前言：
上一篇我们分享了 Kafka 批量消息相关的知识，本篇我们继续分享 Kafka 的广播消费。
Kafka 系列文章传送门
Kafka 客户端工具使用分享【offsetexplorer】
什么是消息广播消费？
传统的消息消费模型有两种，分别是：
- 队列模式：也叫点到点消费，一条消息只会被多个消费端中的一个消费端消费掉。
- 发布订阅模式：消息会被广播给所有的订阅者，也就是只要订阅了的消费端都可以消费同一条消息。
Kafka 基于以上两种模式，提出了一个消费组的概念，即 Consumer Group，一条消息只能被同一个 Consumer Group 中的某个消费端消费，但同一条消息可以被不同消费组中的消费端来消费，Kafka 的广播消费就是依赖 Consumer Group 来实现的。
消费组 Consumer Group 的定义一般会有两种方式来设定，如下：
- 使用业务属性直接定义消费组，也就是人为来设定消费组保证每个类型的业务的消费组不重复。
- 使用 Spring EL 表达式，在每个消费者分组 groupId 上加上 UUID，这样就能保证每个项目启动的消费者分组不同，从而达到广播消费的目的。
广播消息的应用场景
广播消息一般用于以下场景：
- 消息推送：例如用户下单后，需要将下单信息推送到库存系统、物流系统、积分系统等。
- 缓存同步：高并发分布式系统中，很多服务都是使用的本地缓存，来提升系统性能，当缓存的对象发生变化时候，使用广播消息，每个节点都会消费消息，更新本地缓存。
Kafka 广播消息案例演示
Kafak 消息 Producer
Kafka 的消息发送没有什么特殊之处，按自己的需求来调用 API 即可，案例代码如下：
package com.order.service.kafka.producer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.Date;
/**
* @ClassName： MyKafkaProducer
* @Author: Author
* @Date: 2024/10/22 19:22
* @Description:
*/
@Slf4j
@Component
public class MyKafkaProducer {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void sendMessage( String message) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String dateStr = sdf.format(new Date());
this.kafkaTemplate.send("my-topic", message);
log.info("完成消息发送,当前时间:{}",dateStr);
}
}
Kafak 消息 Consumer
因为我们演示的是 Kafka 的广播消息消费，因此我们需要使用不同的消费组和不同的消费端，这里我们使用两个消费端来实现。
消费端一：
package com.order.service.kafka.consumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.Date;
/**
* @ClassName： MyKafkaConsumer
* @Author: Author
* @Date: 2024/10/22 19:22
* @Description:
*/
@Slf4j
@Component
public class MyKafkaConsumer {
@KafkaListener(id = "my-kafka-consumer",
groupId = "my-kafka-consumer-groupId",
topics = "my-topic",
containerFactory = "myContainerFactory")
public void listen(String message) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String dateStr = sdf.format(new Date());
log.info("消息消费成功,当前时间:{},消息内容:{}", dateStr, message);
}
}
消费端一我们指定了消费组，当然也可以不指定，在配置文件中配置全局消费组。
消费端二：
package com.order.service.kafka.consumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
/**
* @ClassName： MyKafkaBroadCastConsumer
* @Author: zhangyong
* @Date: 2024/10/25 17:39
* @Description:
*/
@Slf4j
@Component
public class MyKafkaBroadCastConsumer {
@KafkaListener(id = "my-kafka-consumer-02",
groupId = "my-kafka-consumer-groupId-#{T(java.util.UUID).randomUUID()})",
topics = "my-topic",
containerFactory = "myContainerFactory")
public void listen(String message) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String dateStr = sdf.format(new Date());
log.info("my-kafka-consumer-groupId-02 消息消费成功,当前时间:{},消息内容:{}", dateStr, message);
}
}
消费端二的 Consumer Group 我们使用 Spring EL 表达式，在每个 groupId 上配合 UUID 生成其后缀 UUID 来实现，保证了消费组不重复，来实现广播消费的效果。
Kafak 消息广播消费结果验证
我们触发消息发送得到如下结果：
2024-10-27 16:42:35.272 INFO 26972 --- [nio-8086-exec-2] c.o.s.kafka.producer.MyKafkaProducer : 完成消息发送,当前时间:2024-10-27 16:42:35
2024-10-27 16:42:35.378 INFO 26972 --- [nsumer-02-0-C-1] c.o.s.k.c.MyKafkaBroadCastConsumer : my-kafka-consumer-groupId-02 消息消费成功,当前时间:2024-10-27 16:42:35,消息内容:第一条 kafka 消息
2024-10-27 16:42:35.378 INFO 26972 --- [-consumer-0-C-1] c.o.s.kafka.consumer.MyKafkaConsumer : 消息消费成功,当前时间:2024-10-27 16:42:35,消息内容:第一条 kafka 消息
可以看到一条消息被消费了两次，也就是两个消费端都消费到了消息，结果符合预期。
如有不正确的地方欢迎各位指出纠正。