文章目錄

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

---

概述

傳統的消息傳遞模式有2種：隊列( queue) 和（publish-subscribe）

• queue模式：多個consumer從服務器中讀取數據，消息只會到達一個consumer
• publish-subscribe模式：消息會被廣播給所有的consumer

Kafka基于這2種模式提供了一種consumer的抽象概念： consumer group

• queue模式：所有的consumer都位于同一個consumer group 下。
• publish-subscribe模式：所有的consumer都有著自己唯一的consumer group

說明： 由2個broker組成的kafka集群，某個主題總共有4個partition(P0-P3)，分別位于不同的broker上。這個集群由2個Consumer Group消費， A有2個consumer instances ，B有4個。

通常一個topic會有幾個consumer group，每個consumer group都是一個邏輯上的訂閱者（ logicalsubscriber ）。每個consumer group由多個consumer instance組成，從而達到可擴展和容災的功能。

---

廣播模式的應用 ----> 應用里緩存了數據字典等配置表在內存中，可以通過 Kafka 廣播消費，實現每個應用節點都消費消息，刷新本地內存的緩存。

---

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: latest # 在廣播訂閱下，一般情況下，無需消費歷史的消息，而是從訂閱的 Topic 的隊列的尾部開始消費即可key-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

auto-offset-reset: latest 廣播模式，一般情況下，無需消費歷史的消息，從訂閱的 Topic 的隊列的尾部開始消費即可

---

生產者

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 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 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 + "-" + "#{T(java.util.UUID).randomUUID()})")public void onMessage(MessageMock messageMock){logger.info("【接受到消息][線程:{} 消息內容：{}]", Thread.currentThread().getName(), messageMock);}}

注意: groupId 通過 Spring EL 表達式，在每個消費者分組的名字上配合 UUID 生成其后綴。這樣，就能保證每個項目啟動的消費者分組不同，從而達到廣播消費的目的。

---

單元測試

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("開始發送 測試廣播模式");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());}});// 阻塞等待，保證消費new CountDownLatch(1).await();}}

啟動多次單元測試， 觀察消息的接受情況

---

測速結果

可以看到不消費組下的 消費者（目前是一個消費組下一個消費者） 均收到了 這條消息，這就是廣播模式

---

源碼地址

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

總結

以上是生活随笔為你收集整理的Apache Kafka-通过设置Consumer Group实现广播模式的全部內容，希望文章能夠幫你解決所遇到的問題。

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