記錄：465

場(chǎng)景：一個(gè)Producer在一個(gè)Topic發(fā)布消息，多個(gè)消費(fèi)者Consumer訂閱Kafka的Topic。每個(gè)Consumer指定一個(gè)特定的ConsumerGroup，達(dá)到一條消息被多個(gè)不同的ConsumerGroup消費(fèi)。

版本：JDK 1.8,Spring?Boot 2.6.3,kafka_2.12-2.8.0,spring-kafka-2.8.2。

Kafka集群安裝：https://blog.csdn.net/zhangbeizhen18/article/details/131156084

1.基礎(chǔ)概念

Topic：Kafka根據(jù)Topic對(duì)消息進(jìn)行歸類，發(fā)布到Kafka的每條消息都需要指定一個(gè)Topic。

Producer：消息生產(chǎn)者，向Broker發(fā)送消息的客戶端。

Consumer：消息消費(fèi)者，從Broker讀取消息的客戶端。

ConsumerGroup：每個(gè)Consumer屬于一個(gè)特定的ConsumerGroup，一條消息可以被多個(gè)不同的ConsumerGroup消費(fèi)；但是一個(gè)ConsumerGroup中只能有一個(gè)Consumer能夠消費(fèi)該消息。

publish：發(fā)布，使用Producer向Kafka寫入數(shù)據(jù)。

subscribe：訂閱，使用Consumer從Kafka讀取數(shù)據(jù)。

2.微服務(wù)中配置Kafka信息

(1)在pom.xml添加依賴

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
  <version>2.8.2</version>
</dependency>

請(qǐng)知悉：spring-kafka框架底層使用了原生的kafka-clients。本例對(duì)應(yīng)版本：3.0.0。

(2)在application.yml中配置Kafka信息

配置參考官網(wǎng)的configuration：https://kafka.apache.org/documentation/

(1)application.yml配置內(nèi)容

spring:
  kafka:
    #kafka集群的IP和端口,格式:(ip:port)
    bootstrap-servers:
      - 192.168.19.161:29092
      - 192.168.19.162:29092
      - 192.168.19.163:29092
    #生產(chǎn)者
    producer:
      #客戶端發(fā)送服務(wù)端失敗的重試次數(shù)
      retries: 2
      #多個(gè)記錄被發(fā)送到同一個(gè)分區(qū)時(shí),生產(chǎn)者將嘗試將記錄一起批處理成更少的請(qǐng)求.
      #此設(shè)置有助于提高客戶端和服務(wù)器的性能,配置控制默認(rèn)批量大小(以字節(jié)為單位)
      batch-size: 16384
      #生產(chǎn)者可用于緩沖等待發(fā)送到服務(wù)器的記錄的總內(nèi)存字節(jié)數(shù)(以字節(jié)為單位)
      buffer-memory: 33554432
      #指定key使用的序列化類
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      #指定value使用的序列化類
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      #生產(chǎn)者producer要求leader節(jié)點(diǎn)在考慮完成請(qǐng)求之前收到的確認(rèn)數(shù),用于控制發(fā)送記錄在服務(wù)端的持久化
      #acks=0,設(shè)置為0,則生產(chǎn)者producer將不會(huì)等待來(lái)自服務(wù)器的任何確認(rèn).該記錄將立即添加到套接字(socket)緩沖區(qū)并視為已發(fā)送.在這種情況下,無(wú)法保證服務(wù)器已收到記錄,并且重試配置(retries)將不會(huì)生效(因?yàn)榭蛻舳送ǔ２粫?huì)知道任何故障),每條記錄返回的偏移量始終設(shè)置為-1.
      #acks=1,設(shè)置為1,leader節(jié)點(diǎn)會(huì)把記錄寫入本地日志,不需要等待所有follower節(jié)點(diǎn)完全確認(rèn)就會(huì)立即應(yīng)答producer.在這種情況下,在follower節(jié)點(diǎn)復(fù)制前,leader節(jié)點(diǎn)確認(rèn)記錄后立即失敗的話,記錄將會(huì)丟失.
      #acks=all,acks=-1,leader節(jié)點(diǎn)將等待所有同步復(fù)制副本完成再確認(rèn)記錄,這保證了只要至少有一個(gè)同步復(fù)制副本存活,記錄就不會(huì)丟失.
      acks: -1
    consumer:
      #開啟consumer的偏移量(offset)自動(dòng)提交到Kafka
      enable-auto-commit: true
      #consumer的偏移量(offset)自動(dòng)提交的時(shí)間間隔,單位毫秒
      auto-commit-interval: 1000
      #在Kafka中沒有初始化偏移量或者當(dāng)前偏移量不存在情況
      #earliest,在偏移量無(wú)效的情況下,自動(dòng)重置為最早的偏移量
      #latest,在偏移量無(wú)效的情況下,自動(dòng)重置為最新的偏移量
      #none,在偏移量無(wú)效的情況下,拋出異常.
      auto-offset-reset: latest
      #一次調(diào)用poll返回的最大記錄條數(shù)
      max-poll-records: 500
      #請(qǐng)求阻塞的最大時(shí)間(毫秒)
      fetch-max-wait: 500
      #請(qǐng)求應(yīng)答的最小字節(jié)數(shù)
      fetch-min-size: 1
      #心跳間隔時(shí)間(毫秒)
      heartbeat-interval: 3000
      #指定key使用的反序列化類
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      #指定value使用的反序列化類
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

(2)解析

配置類在spring boot自動(dòng)注解包：spring-boot-autoconfigure-2.6.3.jar。

類：org.springframework.boot.autoconfigure.kafka.KafkaProperties。

使用@ConfigurationProperties注解使其生效，前綴是：spring.kafka。

spring-kafka框架對(duì)操作Kafka單機(jī)版和Kafka集群版的配置差異：

在于bootstrap-servers屬性，單機(jī)版配置一個(gè)IP:端口對(duì)。集群版配置多個(gè)IP:端口對(duì)就行。

(3)加載邏輯

Spring Boot微服務(wù)在啟動(dòng)時(shí)，Spring Boot會(huì)讀取application.yml的配置信息，根據(jù)配置內(nèi)容在spring-boot-autoconfigure-2.6.3.jar找到KafkaProperties并注入到對(duì)應(yīng)屬性。Spring Boot微服務(wù)在啟動(dòng)完成后，KafkaProperties的配置信息在Spring環(huán)境中就能無(wú)縫使用。

Spring的spring-kafka框架將KafkaProperties配置信息注入到KafkaTemplate操作生產(chǎn)者Producer。

Spring的spring-kafka框架使用KafkaProperties和@KafkaListener操作Kafka的消費(fèi)者Consumer。

3.生產(chǎn)者(ChangjiangDeltaCityProducerController)

(1)示例代碼

@RestController
@RequestMapping("/hub/example/delta/producer")
@Slf4j
public class ChangjiangDeltaCityProducerController {
  //1.注入KafkaTemplate
  @Autowired
  private KafkaTemplate<String, String> kafkaTemplate;
  //2.定義Kafka的Topic
  private final String topicName = "hub-topic-city-delta";
  @GetMapping("/f01_1")
  public Object f01_1(String msgContent) {
    try {
      //3.獲取業(yè)務(wù)數(shù)據(jù)對(duì)象
      String uuid=UUID.randomUUID().toString().replace("-","");
      long now=System.currentTimeMillis();
      String msgKey = "delta" + ":" + uuid + ":" + now;
      MsgDto msgDto = MsgDto.buildDto(uuid,now,msgContent);
      String msgData = JSONObject.toJSONString(msgDto);
      log.info("KafkaProducer向Kafka集群的Topic: {},寫入Key:", topicName);
      log.info(msgKey);
      log.info("KafkaProducer向Kafka集群的Topic: {},寫入Data:", topicName);
      log.info(msgData);
      //4.使用KafkaTemplate向Kafka集群寫入數(shù)據(jù)(topic,key,data)
      kafkaTemplate.send(topicName, msgKey, msgData);
    } catch (Exception e) {
      log.info("Producer寫入Topic異常.");
      e.printStackTrace();
    }
    return "寫入成功";
  }
}

(2)解析代碼

使用KafkaTemplate向Kafka集群的Topic：hub-topic-city-delta寫入JSON字符串?dāng)?shù)據(jù)，發(fā)布一條消息，給訂閱的消費(fèi)者消費(fèi)。

4.消費(fèi)者一(HangzhouCityConsumer)

(1)示例代碼

@Component
@Slf4j
public class HangzhouCityConsumer {
  // 1.定義Kafka的Topic
  private final String topicName = "hub-topic-city-delta";
  // 2.使用@KafkaListener監(jiān)聽Kafka集群的Topic
  @KafkaListener(
      topics = {topicName},
      groupId = "hub-topic-city-delta-group-hangzhou")
  public void consumeMsg(ConsumerRecord<?, ?> record) {
    try {
        //3.KafkaConsumer從集群中監(jiān)聽的消息存儲(chǔ)在ConsumerRecord
        String msgKey= (String) record.key();
        String msgData = (String) record.value();
        log.info("HangzhouCityConsumer從Kafka集群中的Topic:{},消費(fèi)的原始數(shù)據(jù)的Key:",topicName);
        log.info(msgKey);
        log.info("HangzhouCityConsumer從Kafka集群中的Topic:{},消費(fèi)的原始數(shù)據(jù)的Data:",topicName);
        log.info(msgData);
    } catch (Exception e) {
        log.info("HangzhouCityConsumer消費(fèi)Topic異常.");
        e.printStackTrace();
    }
  }
}

(2)解析代碼

使用@KafkaListener的屬性topics指定監(jiān)聽的Topic：hub-topic-city-delta。

使用@KafkaListener的屬性groupId 指定消費(fèi)組：hub-topic-city-delta-group-hangzhou。

5.消費(fèi)者二(ShanghaiCityConsumer)

(1)示例代碼

@Component
@Slf4j
public class ShanghaiCityConsumer {
  // 1.定義Kafka的Topic
  private final String topicName = "hub-topic-city-delta";
  // 2.使用@KafkaListener監(jiān)聽Kafka集群的Topic
  @KafkaListener(
          topics = {topicName},
          groupId = "hub-topic-city-delta-group-shanghai")
  public void consumeMsg(ConsumerRecord<?, ?> record) {
    try {
        //3.KafkaConsumer從集群中監(jiān)聽的消息存儲(chǔ)在ConsumerRecord
        String msgKey = (String) record.key();
        String msgData = (String) record.value();
        log.info("ShanghaiCityConsumer從Kafka集群中的Topic:{},消費(fèi)的原始數(shù)據(jù)的Key:", topicName);
        log.info(msgKey);
        log.info("ShanghaiCityConsumer從Kafka集群中的Topic:{},消費(fèi)的原始數(shù)據(jù)的Data:", topicName);
        log.info(msgData);
    } catch (Exception e) {
        log.info("ShanghaiCityConsumer消費(fèi)Topic異常.");
        e.printStackTrace();
    }
  }
}

(2)解析代碼

使用@KafkaListener的屬性topics指定監(jiān)聽的Topic：hub-topic-city-delta。

使用@KafkaListener的屬性groupId 指定消費(fèi)組：hub-topic-city-delta-group-shanghai。

6.測(cè)試

(1)使用Postman測(cè)試，調(diào)用生產(chǎn)者寫入數(shù)據(jù)

請(qǐng)求RUL：http://127.0.0.1:18208/hub-208-kafka/hub/example/delta/producer/f01_1

參數(shù)：msgContent="長(zhǎng)三角經(jīng)濟(jì)帶實(shí)力強(qiáng)大"

(2)生產(chǎn)者日志

KafkaProducer向Kafka集群的Topic: hub-topic-city-delta,寫入Key:
delta:b5a669933f4041588d53d53c22888943:1687789723647
KafkaProducer向Kafka集群的Topic: hub-topic-city-delta,寫入Data:
{"msgContent":"長(zhǎng)三角經(jīng)濟(jì)帶實(shí)力強(qiáng)大","publicTime":"2023-06-26 22:28:43","uuid":"b5a669933f4041588d53d53c22888943"}

(3)消費(fèi)者一日志

HangzhouCityConsumer從Kafka集群中的Topic:hub-topic-city-delta,消費(fèi)的原始數(shù)據(jù)的Key:
delta:b5a669933f4041588d53d53c22888943:1687789723647
HangzhouCityConsumer從Kafka集群中的Topic:hub-topic-city-delta,消費(fèi)的原始數(shù)據(jù)的Data:
{"msgContent":"長(zhǎng)三角經(jīng)濟(jì)帶實(shí)力強(qiáng)大","publicTime":"2023-06-26 22:28:43","uuid":"b5a669933f4041588d53d53c22888943"}

(4)消費(fèi)者二日志

ShanghaiCityConsumer從Kafka集群中的Topic:hub-topic-city-delta,消費(fèi)的原始數(shù)據(jù)的Key:
delta:b5a669933f4041588d53d53c22888943:1687789723647
ShanghaiCityConsumer從Kafka集群中的Topic:hub-topic-city-delta,消費(fèi)的原始數(shù)據(jù)的Data:
{"msgContent":"長(zhǎng)三角經(jīng)濟(jì)帶實(shí)力強(qiáng)大","publicTime":"2023-06-26 22:28:43","uuid":"b5a669933f4041588d53d53c22888943"}

(5)結(jié)論

每個(gè)Consumer指定一個(gè)特定的ConsumerGroup，一條消息可以被多個(gè)不同的ConsumerGroup消費(fèi)。

7.輔助類

@Data
@Builder
public class MsgDto implements Serializable {
  private String uuid;
  private String publicTime;
  private String msgContent;
  public static MsgDto buildDto(String uuid,
                      long publicTime,
                      String msgContent) {
      return builder().uuid(uuid)
          .publicTime(DateUtil.formatDateTime(new Date(publicTime)))
          .msgContent(msgContent).build();
  }
}

以上，感謝。

2023年6月26日文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-552039.html

到了這里，關(guān)于多個(gè)消費(fèi)者訂閱一個(gè)Kafka的Topic(使用@KafkaListener和KafkaTemplate)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！