快捷導(dǎo)航

SpringBoot如何正確配置并運(yùn)行Kafka

更新時間：2023年07月19日 09:55:32 作者：學(xué)弟不想努力了

這篇文章主要介紹了SpringBoot如何正確配置并運(yùn)行Kafka問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

一、配置pom.xml，引入maven依賴

<!-- 引入kafka依賴 -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.8.6</version>
</dependency>

二、application.yml配置文件

這里只提供了kafka有用的相關(guān)配置，其他的配置刪了

spring:
  kafka:
    bootstrap-servers: xx.xx.xx.xx:9092 # kafka集群信息，多個用逗號間隔
    # 生產(chǎn)者
    producer:
      # 重試次數(shù)，設(shè)置大于0的值，則客戶端會將發(fā)送失敗的記錄重新發(fā)送
      retries: 3
      batch-size: 16384 #批量處理大小，16K
      buffer-memory: 33554432 #緩沖存儲大，32M
      acks: 1
      # 指定消息key和消息體的編解碼方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    # 消費(fèi)者
    consumer:
      # 消費(fèi)者組
      group-id: TestGroup
      # 是否自動提交
      enable-auto-commit: false
      # 消費(fèi)偏移配置
      # none：如果沒有為消費(fèi)者找到先前的offset的值,即沒有自動維護(hù)偏移量,也沒有手動維護(hù)偏移量,則拋出異常
      # earliest：在各分區(qū)下有提交的offset時：從offset處開始消費(fèi)；在各分區(qū)下無提交的offset時：從頭開始消費(fèi)
      # latest：在各分區(qū)下有提交的offset時：從offset處開始消費(fèi)；在各分區(qū)下無提交的offset時：從最新的數(shù)據(jù)開始消費(fèi)
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    # 監(jiān)聽
    listener:
      # record：當(dāng)每一條記錄被消費(fèi)者監(jiān)聽器（ListenerConsumer）處理之后提交
      # batch：當(dāng)每一批poll()的數(shù)據(jù)被ListenerConsumer處理之后提交
      # time：當(dāng)每一批poll()的數(shù)據(jù)被ListenerConsumer處理之后，距離上次提交時間大于TIME時提交
      # count：當(dāng)每一批poll()的數(shù)據(jù)被ListenerConsumer處理之后，被處理record數(shù)量大于等于COUNT時提交
      # count_time：TIME或COUNT中有一個條件滿足時提交
      # manual：當(dāng)每一批poll()的數(shù)據(jù)被ListenerConsumer處理之后, 手動調(diào)用Acknowledgment.acknowledge()后提交
      # manual_immediate：手動調(diào)用Acknowledgment.acknowledge()后立即提交，一般推薦使用這種
      ack-mode: manual_immediate

三、消費(fèi)者

消費(fèi)者監(jiān)聽，可以配置多個監(jiān)聽器

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
/**
 * 消費(fèi)者
 * kafka監(jiān)聽器
 */
@Component
public class KafkaConsumer {
    /**
     * kafka的監(jiān)聽器1，topic為"topic_test"，消費(fèi)者組為"group_topic_test"
     * @param record
     * @param item
     */
    @KafkaListener(topics = "topic_test", groupId = "group_topic_test")
    public void topicListener1(ConsumerRecord<String, String> record, Acknowledgment item) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        //手動提交
        item.acknowledge();
    }
    /**
     * 配置多個消費(fèi)組
     * kafka的監(jiān)聽器2，topic為"topic_test2"，消費(fèi)者組為"group_topic_test"
     * @param record
     * @param item
     */
    @KafkaListener(topics = "topic_test2",groupId = "group_topic_test2")
    public void topicListener2(ConsumerRecord<String, String> record, Acknowledgment item) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        item.acknowledge();
    }
}

四、生產(chǎn)者

生產(chǎn)者作為接口Api作為測試

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;
/**
 * kafka生產(chǎn)者
 */
@RestController
@RequestMapping("/kafka")
public class KafkaController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @RequestMapping("/send")
    public void send() {
        kafkaTemplate.send("topic_test",  "key", "測試kafka消息");
    }
}