快捷導(dǎo)航

SpringBoot集成Kafka2.4.0的全過(guò)程

更新時(shí)間：2025年09月29日 09:50:10 作者：云暮冬

本文介紹了如何在Spring Boot 2.4.0環(huán)境中集成Kafka,從前提條件、Kafka啟動(dòng)（包括Windows和Linux）、pom.xml配置、消息生產(chǎn)者與消費(fèi)者代碼編寫到實(shí)際運(yùn)行并發(fā)送消息的全過(guò)程,感興趣的朋友跟隨小編一起看看吧

1、前提條件

前提是需要安裝Kafka及能正常啟動(dòng)，正常啟動(dòng)后記錄下Kafka的ip及端口號(hào)，例如我的：127.0.0.1:9092。
由于用的Kafka還是需要zookeeper的，因此之前也要安裝啟動(dòng)好zookeeper，zookeeper的安裝啟動(dòng)見(jiàn)這篇文章：

zookeeper入門

2、啟動(dòng)Kafka

2.1 windows啟動(dòng)

1、到Kafka安裝目錄下的bin目錄下的windows下：

2、執(zhí)行命令：kafka-server-start.bat -daemon D:\java\kafka_2.11-2.4.0\config\server.properties

2.2 linux啟動(dòng)

3、引入pom

這里由于我系統(tǒng)安裝的是2.4.0版本的Kafka，因此pom引入最好用相同版本的，避免出現(xiàn)其他問(wèn)題。還有一點(diǎn)要注意的是，如果最新的spring boot引入這個(gè)版本zookeeper可能有問(wèn)題，因此需要注意。我用的是2.3.7.RELEASE版本的spring boot。

<dependency>
  <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.0</version>
</dependency>

4、編寫配置文件

這里一些相應(yīng)的配置后續(xù)可以專門更新一篇文章，目前的目的是先跑起來(lái)體驗(yàn)一下。搭建起來(lái)。

spring:
  application:
    name: xuydkafka
  kafka:
    bootstrap-servers: 127.0.0.1:9092 #kafka地址
    producer: # 生產(chǎn)者
      retries: 3 # 設(shè)置大于0的值，則客戶端會(huì)將發(fā)送失敗的記錄重新發(fā)送
      batch-size: 16384
      buffer-memory: 33554432
      acks: 1
      # 指定消息key和消息體的編解碼方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: default-group
      enable-auto-commit: false
      auto-offset-reset: earliest # 從消息頭開始poll()數(shù)據(jù)
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      max-poll-records: 500 # 一次poll最大的消息數(shù)量
    listener:
      # 當(dāng)每一條記錄被消費(fèi)者監(jiān)聽(tīng)器（ListenerConsumer）處理之后提交
      # RECORD
      # 當(dāng)每一批poll()的數(shù)據(jù)被消費(fèi)者監(jiān)聽(tīng)器（ListenerConsumer）處理之后提交
      # BATCH
      # 當(dāng)每一批poll()的數(shù)據(jù)被消費(fèi)者監(jiān)聽(tīng)器（ListenerConsumer）處理之后，距離上次提交的時(shí)間大于time時(shí)提交
      # TIME
      # 當(dāng)每一批poll()的數(shù)據(jù)被消費(fèi)者監(jiān)聽(tīng)器（ListenerConsumer）處理之后，被處理的record數(shù)量大于等于count時(shí)提交
      # COUNT
      # TIME | COUNT 有?個(gè)條件滿?時(shí)提交
      # COUNT_TIME
      # 當(dāng)每?批poll()的數(shù)據(jù)被消費(fèi)者監(jiān)聽(tīng)器（ListenerConsumer）處理之后, ?動(dòng)調(diào)#?Acknowledgment.acknowledge()后提交
      # MANUAL
      #  ?動(dòng)調(diào)?Acknowledgment.acknowledge()后?即提交，?般使?這種
      # MANUAL_IMMEDIATE
      ack-mode: MANUAL_IMMEDIATE
  redis:
    host: 127.0.0.1
server:
  port: 9000

5、編寫消息生產(chǎn)者代碼

package com.xuyd.kafka.controller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/msg")
public class KafkaController {
    private final static String TOPIC_NAME = "my-replicated-topic";
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @RequestMapping("/send")
    public String sendMessage(){
        kafkaTemplate.send(TOPIC_NAME, 0, "key", "this is a message");
        return "send success";
    }
}

6、編寫消息消費(fèi)者相關(guān)代碼

需要注意的是，一般生產(chǎn)者和消費(fèi)者不會(huì)在同一個(gè)項(xiàng)目。這里只是體驗(yàn)，所以臨時(shí)這么處理。

package com.xuyd.kafka.consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
@Component
public class MyConsumer {
    @KafkaListener(topics = "my-replicated-topic", groupId = "MyGroup1")
    public void listenGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        // 由于配置的提交方式是MANUAL_IMMEDIATE 因此這里需要手動(dòng)提交代碼        
        ack.acknowledge();
    }
}