文章目錄

• 概述
• 參數設置
• Code
• • POM依賴
  • 配置文件
  • 生產者
  • 消費者
  • 單元測試
  • 測試結果
• 源碼地址

---

概述

kafka中有個 micro batch 的概念 ，為了提高Producer 發送的性能。

不同于RocketMQ 提供了一個可以批量發送多條消息的 API 。 Kafka 的做法是：提供了一個 RecordAccumulator 消息收集器，將發送給相同 Topic 的相同 Partition 分區的消息們，緩沖一下，當滿足條件時候，一次性批量將緩沖的消息提交給 Kafka Broker 。

---

參數設置

https://kafka.apache.org/24/documentation.html#producerconfigs

主要涉及的參數 ，三個條件，滿足任一即會批量發送：

• batch-size ：超過收集的消息數量的最大量。默認16KB

• buffer-memory ：超過收集的消息占用的最大內存 , 默認32M

• linger.ms ：超過收集的時間的最大等待時長，單位：毫秒。

---

Code

POM依賴

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- 引入 Spring-Kafka 依賴 --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><scope>test</scope></dependency></dependencies>

---

配置文件

spring:# Kafka 配置項，對應 KafkaProperties 配置類kafka:bootstrap-servers: 192.168.126.140:9092 # 指定 Kafka Broker 地址，可以設置多個，以逗號分隔# Kafka Producer 配置項producer:acks: 1 # 0-不應答。1-leader 應答。all-所有 leader 和 follower 應答。retries: 3 # 發送失敗時，重試發送的次數key-serializer: org.apache.kafka.common.serialization.StringSerializer # 消息的 key 的序列化value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化batch-size: 16384 # 每次批量發送消息的最大數量 單位 字節 默認 16Kbuffer-memory: 33554432 # 每次批量發送消息的最大內存 單位 字節 默認 32Mproperties:linger:ms: 10000 # 批處理延遲時間上限。[實際不會配這么長，這里用于測速]這里配置為 10 * 1000 ms 過后，不管是否消息數量是否到達 batch-size 或者消息大小到達 buffer-memory 后，都直接發送一次請求。# Kafka Consumer 配置項consumer:auto-offset-reset: earliest # 設置消費者分組最初的消費進度為 earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.springframework.kafka.support.serializer.JsonDeserializerproperties:spring:json:trusted:packages: com.artisan.springkafka.domain# Kafka Consumer Listener 監聽器配置listener:missing-topics-fatal: false # 消費監聽接口監聽的主題不存在時，默認會報錯。所以通過設置為 false ，解決報錯logging:level:org:springframework:kafka: ERROR # spring-kafkaapache:kafka: ERROR # kafka

---

生產者

package com.artisan.springkafka.producer;import com.artisan.springkafka.constants.TOPIC; import com.artisan.springkafka.domain.MessageMock; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.stereotype.Component; import org.springframework.util.concurrent.ListenableFuture;import java.util.Random; import java.util.concurrent.ExecutionException;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:25* @mark: show me the code , change the world*/@Component public class ArtisanProducerMock {@Autowiredprivate KafkaTemplate<Object,Object> kafkaTemplate ;/*** 同步發送* @return* @throws ExecutionException* @throws InterruptedException*/public SendResult sendMsgSync() throws ExecutionException, InterruptedException {// 模擬發送的消息Integer id = new Random().nextInt(100);MessageMock messageMock = new MessageMock(id,"artisanTestMessage-" + id);// 同步等待return kafkaTemplate.send(TOPIC.TOPIC, messageMock).get();}public ListenableFuture<SendResult<Object, Object>> sendMsgASync() throws ExecutionException, InterruptedException {// 模擬發送的消息Integer id = new Random().nextInt(100);MessageMock messageMock = new MessageMock(id,"messageSendByAsync-" + id);// 異步發送消息ListenableFuture<SendResult<Object, Object>> result = kafkaTemplate.send(TOPIC.TOPIC, messageMock);return result ;}}

---

消費者

package com.artisan.springkafka.consumer;import com.artisan.springkafka.domain.MessageMock; import com.artisan.springkafka.constants.TOPIC; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:33* @mark: show me the code , change the world*/@Component public class ArtisanCosumerMock {private Logger logger = LoggerFactory.getLogger(getClass());private static final String CONSUMER_GROUP_PREFIX = "MOCK-A" ;@KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC)public void onMessage(MessageMock messageMock){logger.info("【接受到消息][線程:{} 消息內容：{}]", Thread.currentThread().getName(), messageMock);}} package com.artisan.springkafka.consumer;import com.artisan.springkafka.domain.MessageMock; import com.artisan.springkafka.constants.TOPIC; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:33* @mark: show me the code , change the world*/@Component public class ArtisanCosumerMockDiffConsumeGroup {private Logger logger = LoggerFactory.getLogger(getClass());private static final String CONSUMER_GROUP_PREFIX = "MOCK-B" ;@KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC)public void onMessage(MessageMock messageMock){logger.info("【接受到消息][線程:{} 消息內容：{}]", Thread.currentThread().getName(), messageMock);}}

---

單元測試

package com.artisan.springkafka.produceTest;import com.artisan.springkafka.SpringkafkaApplication; import com.artisan.springkafka.producer.ArtisanProducerMock; import org.junit.Test; import org.junit.runner.RunWith; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.kafka.support.SendResult; import org.springframework.test.context.junit4.SpringRunner; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback;import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit;/*** @author 小工匠* * @version 1.0* @description: TODO* @date 2021/2/17 22:40* @mark: show me the code , change the world*/@RunWith(SpringRunner.class) @SpringBootTest(classes = SpringkafkaApplication.class) public class ProduceMockTest {private Logger logger = LoggerFactory.getLogger(getClass());@Autowiredprivate ArtisanProducerMock artisanProducerMock;@Testpublic void testAsynSend() throws ExecutionException, InterruptedException {logger.info("開始發送");for (int i = 0; i < 2; i++) {artisanProducerMock.sendMsgASync().addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {@Overridepublic void onFailure(Throwable throwable) {logger.info(" 發送異常{}]]", throwable);}@Overridepublic void onSuccess(SendResult<Object, Object> objectObjectSendResult) {logger.info("回調結果 Result = topic:[{}] , partition:[{}], offset:[{}]",objectObjectSendResult.getRecordMetadata().topic(),objectObjectSendResult.getRecordMetadata().partition(),objectObjectSendResult.getRecordMetadata().offset());}});// 發送2次 每次間隔5秒， 湊夠我們配置的 linger: ms: 10000TimeUnit.SECONDS.sleep(5);}// 阻塞等待，保證消費new CountDownLatch(1).await();}}

異步發送2條消息，每次發送消息之間， sleep 5 秒，以便達到配置的 linger.ms 最大等待時長10秒。

---

測試結果

2021-02-18 10:58:53.360 INFO 24736 --- [ main] c.a.s.produceTest.ProduceMockTest : 開始發送 2021-02-18 10:59:03.555 INFO 24736 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest : 回調結果 Result = topic:[MOCK_TOPIC] , partition:[0], offset:[30] 2021-02-18 10:59:03.556 INFO 24736 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest : 回調結果 Result = topic:[MOCK_TOPIC] , partition:[0], offset:[31] 2021-02-18 10:59:03.595 INFO 24736 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][線程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息內容：MessageMock{id=6, name='messageSendByAsync-6'}] 2021-02-18 10:59:03.595 INFO 24736 --- [ntainer#1-0-C-1] a.s.c.ArtisanCosumerMockDiffConsumeGroup : 【接受到消息][線程:org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1 消息內容：MessageMock{id=6, name='messageSendByAsync-6'}] 2021-02-18 10:59:03.595 INFO 24736 --- [ntainer#1-0-C-1] a.s.c.ArtisanCosumerMockDiffConsumeGroup : 【接受到消息][線程:org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1 消息內容：MessageMock{id=94, name='messageSendByAsync-94'}] 2021-02-18 10:59:03.595 INFO 24736 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][線程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息內容：MessageMock{id=94, name='messageSendByAsync-94'}]

10 秒后，滿足批量消息的最大等待時長，所以 2 條消息被 Producer 批量發送。同時我們配置的是 acks=1 ，需要等待發送成功后，才會回調 ListenableFutureCallback 的方法。

當然了，我們這里都是為了測試，設置的這么長的間隔，實際中需要根據具體的業務場景設置一個合理的值。

---

源碼地址

https://github.com/yangshangwei/boot2/tree/master/springkafkaBatchSend

總結

以上是生活随笔為你收集整理的Apache Kafka-生产者_批量发送消息的核心参数及功能实现的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。