記錄：463

場景：在Spring Boot微服務(wù)集成kafka-clients-3.0.0操作Kafka集群。使用kafka-clients的原生KafkaProducer操作Kafka集群生產(chǎn)者Producer。使用kafka-clients的原生KafkaConsumer操作Kafka集群的消費者Consumer。

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

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

1.基礎(chǔ)概念

Event：An event records the fact that "something happened" in the world or in your business. It is also called record or message in the documentation.

Broker：一個Kafka節(jié)點就是一個broker；多個Broker可以組成一個Kafka集群。

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

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

Consumer：消息消費者，從Broker讀取消息的客戶端。

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

Partition：一個topic可以分為多個partition，每個partition內(nèi)部消息是有序的。

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

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

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

(1)在pom.xml添加依賴

pom.xml文件：

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

解析：使用原生的kafka-clients，版本：3.0.0。操作kafka的生產(chǎn)者、消費、Topic。

3.配置Kafka生產(chǎn)者和消費者

使用原生的kafka-clients，需配置KafkaProducer和KafkaConsumer，把Kafka集群的配置信息注入到這兩個對象，便可以操作了生產(chǎn)者和消費者。

配置細節(jié)在官網(wǎng)的configuration：https://kafka.apache.org/documentation/

3.1配置KafkaProducer生產(chǎn)者

(1)示例代碼

@Configuration
public class KafkaClusterConfig {
  @Bean
  public KafkaProducer kafkaProducer() {
      Map<String, Object> configs = new HashMap<>();
      //kafka集群
      Collection<String> cluster = Lists.newArrayList("192.168.19.161:29092",
              "192.168.19.162:29092",
              "192.168.19.163:29092");
      configs.put("bootstrap.servers", cluster);
      //客戶端發(fā)送服務(wù)端失敗的重試次數(shù)
      configs.put("retries", 2);
      //多個記錄被發(fā)送到同一個分區(qū)時,生產(chǎn)者將嘗試將記錄一起批處理成更少的請求.
      //此設(shè)置有助于提高客戶端和服務(wù)器的性能,配置控制默認批量大小(以字節(jié)為單位)
      configs.put("batch.size", 16384);
      //生產(chǎn)者可用于緩沖等待發(fā)送到服務(wù)器的記錄的總內(nèi)存字節(jié)數(shù)(以字節(jié)為單位)
      configs.put("buffer-memory", 33554432);
      //生產(chǎn)者producer要求leader節(jié)點在考慮完成請求之前收到的確認數(shù),用于控制發(fā)送記錄在服務(wù)端的持久化
      //acks=0,設(shè)置為0,則生產(chǎn)者producer將不會等待來自服務(wù)器的任何確認.該記錄將立即添加到套接字(socket)緩沖區(qū)并視為已發(fā)送.在這種情況下,無法保證服務(wù)器已收到記錄,并且重試配置(retries)將不會生效(因為客戶端通常不會知道任何故障),每條記錄返回的偏移量始終設(shè)置為-1.
      //acks=1,設(shè)置為1,leader節(jié)點會把記錄寫入本地日志,不需要等待所有follower節(jié)點完全確認就會立即應(yīng)答producer.在這種情況下,在follower節(jié)點復(fù)制前,leader節(jié)點確認記錄后立即失敗的話,記錄將會丟失.
      //acks=all,acks=-1,leader節(jié)點將等待所有同步復(fù)制副本完成再確認記錄,這保證了只要至少有一個同步復(fù)制副本存活,記錄就不會丟失.
      configs.put("acks", "-1");
      //指定key使用的序列化類
      Serializer keySerializer = new StringSerializer();
      //指定value使用的序列化類
      Serializer valueSerializer = new StringSerializer();
      //創(chuàng)建Kafka生產(chǎn)者
      KafkaProducer kafkaProducer = new KafkaProducer(configs, keySerializer, valueSerializer);
      return kafkaProducer;
  }
}

(2)解析代碼

把Kafka的配置信息注入到KafkaProducer，并創(chuàng)建KafkaProducer對象。

使用@Configuration和@Bean注解把KafkaProducer對象注入到Spring的IOC容器，在Spring環(huán)境就可以使用KafkaProducer了。

KafkaProducer的底層使用配置類是ProducerConfig，在配置時可以參考。

全稱：org.apache.kafka.clients.producer.ProducerConfig。

3.2配置KafkaConsumer消費者

(1)示例代碼

@Configuration
public class KafkaClusterConfig {
  @Bean
  public KafkaConsumer kafkaConsumer() {
      Map<String, Object> configs = new HashMap<>();
      //kafka集群
      Collection<String> cluster = Lists.newArrayList("192.168.19.161:29092",
              "192.168.19.162:29092",
              "192.168.19.163:29092");
      configs.put("bootstrap.servers", cluster);
      //開啟consumer的偏移量(offset)自動提交到Kafka
      configs.put("enable.auto.commit", true);
      //consumer的偏移量(offset) 自動提交的時間間隔,單位毫秒
      configs.put("auto.commit.interval", 5000);
      //在Kafka中沒有初始化偏移量或者當前偏移量不存在情況
      //earliest, 在偏移量無效的情況下, 自動重置為最早的偏移量
      //latest, 在偏移量無效的情況下, 自動重置為最新的偏移量
      //none, 在偏移量無效的情況下, 拋出異常.
      configs.put("auto.offset.reset", "latest");
      //請求阻塞的最大時間(毫秒)
      configs.put("fetch.max.wait", 500);
      //請求應(yīng)答的最小字節(jié)數(shù)
      configs.put("fetch.min.size", 1);
      //心跳間隔時間(毫秒)
      configs.put("heartbeat-interval", 3000);
      //一次調(diào)用poll返回的最大記錄條數(shù)
      configs.put("max.poll.records", 500);
      //指定消費組
      configs.put("group.id", "hub-topic-city-info-02-group");
      //指定key使用的反序列化類
      Deserializer keyDeserializer = new StringDeserializer();
      //指定value使用的反序列化類
      Deserializer valueDeserializer = new StringDeserializer();
      //創(chuàng)建Kafka消費者
      KafkaConsumer kafkaConsumer = new KafkaConsumer(configs, keyDeserializer, valueDeserializer);
      return kafkaConsumer;
  }
}

(2)解析代碼

把Kafka的配置信息注入到KafkaConsumer，并創(chuàng)建KafkaConsumer對象。

使用@Configuration和@Bean注解把KafkaConsumer對象注入到Spring的IOC容器，在Spring環(huán)境就可以使用KafkaConsumer了。

KafkaConsumer的底層使用配置類是ConsumerConfig，在配置時可以參考。

全稱：org.apache.kafka.clients.consumer.ConsumerConfig。

4.使用原生KafkaProducer操作Kafka集群生產(chǎn)者Producer

使用原生kafka-clients的KafkaProducer操作Kafka集群生產(chǎn)者Producer。

KafkaProducer全稱：org.apache.kafka.clients.producer.KafkaProducer。

(1)示例代碼

@RestController
@RequestMapping("/hub/example/cluster/producer")
@Slf4j
public class UseKafkaClusterProducerController {
  //注入Kafka生產(chǎn)者
  @Autowired
  private KafkaProducer kafkaProducer;
  //定義Kafka的Topic
  private final String topicName = "hub-topic-city-info-02";
  @GetMapping("/f01_1")
  public Object f01_1() {
      try {
          //1.獲取業(yè)務(wù)數(shù)據(jù)
          Long cityId = 2023061801L;
          String cityName = "上海";
          String msgKey = cityId + ":" + cityName + ":" + System.currentTimeMillis();
          CityDTO cityDTO = CityDTO.buildDto(cityId, cityName, "上海是一個國際大都市");
          String msgData = JSONObject.toJSONString(cityDTO);
          log.info("KafkaProducer向Kafka集群的Topic: {},寫入Key:",topicName);
          log.info(msgKey);
          log.info("KafkaProducer向Kafka集群的Topic: {},寫入Data:",topicName);
          log.info(msgData);
          //2.使用KafkaProducer向Kafka集群寫入數(shù)據(jù)
          ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topicName, msgKey,msgData);
          kafkaProducer.send(producerRecord);
      } catch (Exception e) {
          log.info("Producer寫入Topic異常.");
          e.printStackTrace();
      }
      return "寫入成功";
  }
}

(2)解析代碼

創(chuàng)建ProducerRecord對象，指定Kafka的Topic名稱、key值、需要寫入的數(shù)data據(jù)，就能完成Producer向Kafka集群的Broker節(jié)點寫入數(shù)據(jù)。ProducerRecord就是需要寫入Kafka中的一條數(shù)據(jù)，

使用KafkaProducer 的send方法，傳入ProducerRecord，就能完成Producer向Kafka的Broker節(jié)點寫入數(shù)據(jù)。

5.使用原生KafkaConsumer操作Kafka集群的消費者Consumer

使用原生kafka-clients的KafkaConsumer操作Kafka集群生產(chǎn)者Consumer。

KafkaConsumer全稱：org.apache.kafka.clients.consumer.KafkaConsumer。

(1)示例代碼

@Component
@Slf4j
public class UseKafkaClusterConsumer implements CommandLineRunner {
  //注入Kafka消費者
  @Autowired
  private KafkaConsumer kafkaConsumer;
  //定義Kafka的Topic
  private final String topicName = "hub-topic-city-info-02";
  @Override
  public void run(String... args) throws Exception {
      Thread thread = new Thread(new ThreadRunnable());
      thread.start();
  }
  //在線程中使用KafkaConsumer實時監(jiān)聽Kafka集群
  public class ThreadRunnable implements Runnable {
      @Override
      public void run() {
          log.info("啟動線程監(jiān)聽Kafka集群的Topic: {}", topicName);
          ThreadUtil.sleep(500);
          Collection<String> topics = Lists.newArrayList(topicName);
          //1.使用KafkaConsumer在Kafka集群訂閱Topic
          kafkaConsumer.subscribe(topics);
          while (true) {
              //2.使用KafkaConsumer的poll按照指定周期輪詢Kafka集群指定Topic的消息
              ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));
              //3.遍歷從Kafka集群中讀取數(shù)據(jù)集ConsumerRecords
              for (ConsumerRecord consumerRecord : consumerRecords) {
                  //4.從ConsumerRecord中取出消費數(shù)據(jù)
                  String msgKey = (String) consumerRecord.key();
                  String msgData = (String) consumerRecord.value();
                  log.info("KafkaConsumer從Kafka集群中的Topic:{},消費的原始數(shù)據(jù)的Key:", topicName);
                  log.info(msgKey);
                  log.info("KafkaConsumer從Kafka集群中的Topic:{},消費的原始數(shù)據(jù)的Data:", topicName);
                  log.info(msgData);
              }
          }
      }
  }
}

(2)解析代碼

使用?while (true)實時遍歷KafkaConsumer消費者，實時監(jiān)聽Kafka集群消息。

使用KafkaConsumer的subscribe方法訂閱需要監(jiān)聽的Kafka集群的Topic。

使用KafkaConsumer的poll方法輪詢消費者獲取消費消息ConsumerRecord。

從ConsumerRecord中獲取具體消費的業(yè)務(wù)數(shù)據(jù)。

6.測試

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

請求RUL：http://127.0.0.1:18210/hub-210-kafka/hub/example/cluster/producer/f01_1

(2)消費者自動消費數(shù)據(jù)

(3)日志信息

KafkaProducer向Kafka集群的Topic: hub-topic-city-info-02,寫入Key:
2023061801:上海:1687075134616
KafkaProducer向Kafka集群的Topic: hub-topic-city-info-02,寫入Data:
{"cityDescribe":"上海是一個國際大都市","cityId":2023061801,"cityName":"上海","updateTime":"2023-06-18 15:58:54"}
KafkaConsumer從Kafka集群中的Topic:hub-topic-city-info-02,消費的原始數(shù)據(jù)的Key:
2023061801:上海:1687075134616
KafkaConsumer從Kafka集群中的Topic:hub-topic-city-info-02,消費的原始數(shù)據(jù)的Data:
{"cityDescribe":"上海是一個國際大都市","cityId":2023061801,"cityName":"上海","updateTime":"2023-06-18 15:58:54"}

7.輔助類

@Data
@Builder
public class CityDTO {
  private Long cityId;
  private String cityName;
  private String cityDescribe;
  private String updateTime;
  public static CityDTO buildDto(Long cityId, String cityName,
                                 String cityDescribe) {
      return builder().cityId(cityId)
              .cityName(cityName).cityDescribe(cityDescribe)
              .updateTime(DateUtil.formatDateTime(new Date())).build();
  }
}

以上，感謝。

2023年6月18日文章來源地址http://www.zghlxwxcb.cn/news/detail-490006.html

到了這里，關(guān)于在Spring Boot微服務(wù)集成kafka-clients操作Kafka集群的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！