文章目錄

• 概述
• Code
• • POM依賴
  • 配置文件
  • 生產(chǎn)者
  • 消費(fèi)者
  • 單元測試
  • 測試結(jié)果
• 源碼地址

---

概述

一個(gè)partition同一個(gè)時(shí)刻在一個(gè)consumer group中只能有一個(gè)consumer instance在消費(fèi)，從而保證消費(fèi)順序。

consumer group中的consumer instance的數(shù)量不能比一個(gè)Topic中的partition的數(shù)量多，否則，多出來的consumer消費(fèi)不到消息。

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

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

有的時(shí)候，我們需要根據(jù)某個(gè)主鍵來實(shí)現(xiàn)消費(fèi)的有序性，那我們只需要考慮將 Producer 將相關(guān)聯(lián)的消息發(fā)送到 Topic 下的相同的 Partition 即可。 Kafka Producer 發(fā)送消息使用分區(qū)策略的話，根據(jù) key 的哈希值取模來獲取到其在 Topic 下對應(yīng)的 Partition 。 所以我們發(fā)送消息的時(shí)候指定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 配置項(xiàng)，對應(yīng) KafkaProperties 配置類kafka:bootstrap-servers: 192.168.126.140:9092 # 指定 Kafka Broker 地址，可以設(shè)置多個(gè)，以逗號分隔# Kafka Producer 配置項(xiàng)producer:acks: 1 # 0-不應(yīng)答。1-leader 應(yīng)答。all-所有 leader 和 follower 應(yīng)答。retries: 3 # 發(fā)送失敗時(shí)，重試發(fā)送的次數(shù)key-serializer: org.apache.kafka.common.serialization.StringSerializer # 消息的 key 的序列化value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化# Kafka Consumer 配置項(xiàng)consumer:auto-offset-reset: earliest # 設(shè)置消費(fèi)者分組最初的消費(fèi)進(jìn)度為 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 監(jiān)聽器配置listener:missing-topics-fatal: false # 消費(fèi)監(jiān)聽接口監(jiān)聽的主題不存在時(shí)，默認(rèn)會報(bào)錯(cuò)。所以通過設(shè)置為 false ，解決報(bào)錯(cuò)logging:level:org:springframework:kafka: ERROR # spring-kafkaapache:kafka: ERROR # kafka

---

生產(chǎn)者

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

---

消費(fèi)者

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:{} 消息內(nèi)容：{}]", 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("開始發(fā)送");// 模擬發(fā)送多條消息for (int i = 0; i < 5; i++) {SendResult sendResult = artisanProducerMock.sendMsgSync();logger.info("key {} . 分區(qū):{}",sendResult.getProducerRecord().key(),sendResult.getRecordMetadata().partition());SendResult sendResult1 = artisanProducerMock.sendMsgSync2();logger.info("key {} . 分區(qū):{}",sendResult1.getProducerRecord().key(),sendResult1.getRecordMetadata().partition());}// 阻塞等待，保證消費(fèi)new CountDownLatch(1).await();}}

---

測試結(jié)果

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

懂了么，老兄 ~

---

源碼地址

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。