文章目錄

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

---

概述

一個partition同一個時刻在一個consumer group中只能有一個consumer instance在消費，從而保證消費順序。

consumer group中的consumer instance的數量不能比一個Topic中的partition的數量多，否則，多出來的consumer消費不到消息。

Kafka只在partition的范圍內保證消息消費的局部順序性，不能在同一個topic中的多個partition中保證總的消費順序性。

如果有在總體上保證消費順序的需求，那么我們可以通過將topic的partition數量設置為1，將consumer group中的consumer instance數量也設置為1，但是這樣會影響性能，所以kafka的順序消費很少用。

有的時候，我們需要根據某個主鍵來實現消費的有序性，那我們只需要考慮將 Producer 將相關聯的消息發送到 Topic 下的相同的 Partition 即可。 Kafka Producer 發送消息使用分區策略的話，根據 key 的哈希值取模來獲取到其在 Topic 下對應的 Partition 。 所以我們發送消息的時候指定key即可

---

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 的序列化# 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 {// 模擬發送的消息int id = 6687421;MessageMock messageMock = new MessageMock(id,"artisanTestMessage-" + id);// 測試順序消息 以 id 為key (因為使用 String 的方式序列化 key ，所以需要將 id 轉換成 String)return kafkaTemplate.send(TOPIC.TOPIC,String.valueOf(id) ,messageMock).get();}/*** 同步發送* @return* @throws ExecutionException* @throws InterruptedException*/public SendResult sendMsgSync2() throws ExecutionException, InterruptedException {// 模擬發送的消息int id = 1255447;MessageMock messageMock = new MessageMock(id,"artisanTestMessage-" + id);// 測試順序消息 以 id 為key (因為使用 String 的方式序列化 key ，所以需要將 id 轉換成 String)return kafkaTemplate.send(TOPIC.TOPIC,String.valueOf(id) ,messageMock).get();} }

---

消費者

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,concurrency = "2")public void onMessage(MessageMock messageMock){ // logger.info("【接受到消息][線程ID:{} 消息內容：{}]", Thread.currentThread().getId(), 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 < 5; i++) {SendResult sendResult = artisanProducerMock.sendMsgSync();logger.info("key {} . 分區:{}",sendResult.getProducerRecord().key(),sendResult.getRecordMetadata().partition());SendResult sendResult1 = artisanProducerMock.sendMsgSync2();logger.info("key {} . 分區:{}",sendResult1.getProducerRecord().key(),sendResult1.getRecordMetadata().partition());}// 阻塞等待，保證消費new CountDownLatch(1).await();}}

---

測試結果

2021-02-18 23:05:46.903 INFO 24568 --- [ main] c.a.s.produceTest.ProduceMockTest : 開始發送 2021-02-18 23:05:47.088 INFO 24568 --- [ main] c.a.s.produceTest.ProduceMockTest : key 6687421 . 分區:1 2021-02-18 23:05:47.090 INFO 24568 --- [ main] c.a.s.produceTest.ProduceMockTest : key 1255447 . 分區:0 2021-02-18 23:05:47.093 INFO 24568 --- [ main] c.a.s.produceTest.ProduceMockTest : key 6687421 . 分區:1 2021-02-18 23:05:47.096 INFO 24568 --- [ main] c.a.s.produceTest.ProduceMockTest : key 1255447 . 分區:0 2021-02-18 23:05:47.097 INFO 24568 --- [ main] c.a.s.produceTest.ProduceMockTest : key 6687421 . 分區:1 2021-02-18 23:05:47.099 INFO 24568 --- [ main] c.a.s.produceTest.ProduceMockTest : key 1255447 . 分區:0 2021-02-18 23:05:47.101 INFO 24568 --- [ main] c.a.s.produceTest.ProduceMockTest : key 6687421 . 分區:1 2021-02-18 23:05:47.103 INFO 24568 --- [ main] c.a.s.produceTest.ProduceMockTest : key 1255447 . 分區:0 2021-02-18 23:05:47.105 INFO 24568 --- [ main] c.a.s.produceTest.ProduceMockTest : key 6687421 . 分區:1 2021-02-18 23:05:47.106 INFO 24568 --- [ main] c.a.s.produceTest.ProduceMockTest : key 1255447 . 分區:0

懂了么，老兄 ~

---

源碼地址

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

總結

以上是生活随笔為你收集整理的Apache Kafka-消费端_顺序消费的实现的全部內容，希望文章能夠幫你解決所遇到的問題。

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