Kafka 之消息广播消费

前言：

上一篇我们分享了 Kafka 批量消息相关的知识，本篇我们继续分享 Kafka 的广播消费。

Kafka 系列文章传送门

Kafka 简介及核心概念讲解

Spring Boot 整合 Kafka 详解

Kafka @KafkaListener 注解的详解及使用

Kafka 客户端工具使用分享【offsetexplorer】

Kafka 之消息同步/异步发送

Kafka 之批量消息发送消费

什么是消息广播消费？

传统的消息消费模型有两种，分别是：

• 队列模式：也叫点到点消费，一条消息只会被多个消费端中的一个消费端消费掉。
• 发布订阅模式：消息会被广播给所有的订阅者，也就是只要订阅了的消费端都可以消费同一条消息。

Kafka 基于以上两种模式，提出了一个消费组的概念，即 Consumer Group，一条消息只能被同一个 Consumer Group 中的某个消费端消费，但同一条消息可以被不同消费组中的消费端来消费，Kafka 的广播消费就是依赖 Consumer Group 来实现的。

消费组 Consumer Group 的定义一般会有两种方式来设定，如下：

• 使用业务属性直接定义消费组，也就是人为来设定消费组保证每个类型的业务的消费组不重复。
• 使用 Spring EL 表达式，在每个消费者分组 groupId 上加上 UUID，这样就能保证每个项目启动的消费者分组不同，从而达到广播消费的目的。

广播消息的应用场景

广播消息一般用于以下场景：

• 消息推送：例如用户下单后，需要将下单信息推送到库存系统、物流系统、积分系统等。
• 缓存同步：高并发分布式系统中，很多服务都是使用的本地缓存，来提升系统性能，当缓存的对象发生变化时候，使用广播消息，每个节点都会消费消息，更新本地缓存。

Kafka 广播消息案例演示

Kafak 消息 Producer

Kafka 的消息发送没有什么特殊之处，按自己的需求来调用 API 即可，案例代码如下：

packagecom.order.service.kafka.producer;importlombok.extern.slf4j.Slf4j;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.kafka.core.KafkaTemplate;importorg.springframework.stereotype.Component;importjava.text.SimpleDateFormat;importjava.util.Date;/**
 * @ClassName： MyKafkaProducer 
 * @Author: Author
 * @Date: 2024/10/22 19:22
 * @Description:
 */@Slf4j@ComponentpublicclassMyKafkaProducer{@AutowiredprivateKafkaTemplate<String,String> kafkaTemplate;publicvoidsendMessage(String message){SimpleDateFormat sdf =newSimpleDateFormat("yyyy-MM-dd HH:mm:ss");String dateStr = sdf.format(newDate());this.kafkaTemplate.send("my-topic", message);
        log.info("完成消息发送,当前时间:{}",dateStr);}}

Kafak 消息 Consumer

因为我们演示的是 Kafka 的广播消息消费，因此我们需要使用不同的消费组和不同的消费端，这里我们使用两个消费端来实现。

消费端一：

packagecom.order.service.kafka.consumer;importlombok.extern.slf4j.Slf4j;importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.stereotype.Component;importjava.text.SimpleDateFormat;importjava.util.Date;/**
 * @ClassName： MyKafkaConsumer 
 * @Author: Author
 * @Date: 2024/10/22 19:22
 * @Description:
 */@Slf4j@ComponentpublicclassMyKafkaConsumer{@KafkaListener(id ="my-kafka-consumer",
            groupId ="my-kafka-consumer-groupId",
            topics ="my-topic",
            containerFactory ="myContainerFactory")publicvoidlisten(String message){SimpleDateFormat sdf =newSimpleDateFormat("yyyy-MM-dd HH:mm:ss");String dateStr = sdf.format(newDate());
        log.info("消息消费成功,当前时间:{},消息内容:{}", dateStr, message);}}

消费端一我们指定了消费组，当然也可以不指定，在配置文件中配置全局消费组。

消费端二：

packagecom.order.service.kafka.consumer;importlombok.extern.slf4j.Slf4j;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.stereotype.Component;importjava.text.SimpleDateFormat;importjava.util.Date;importjava.util.List;/**
 * @ClassName： MyKafkaBroadCastConsumer
 * @Author: zhangyong
 * @Date: 2024/10/25 17:39
 * @Description:
 */@Slf4j@ComponentpublicclassMyKafkaBroadCastConsumer{@KafkaListener(id ="my-kafka-consumer-02",
            groupId ="my-kafka-consumer-groupId-#{T(java.util.UUID).randomUUID()})",
            topics ="my-topic",
            containerFactory ="myContainerFactory")publicvoidlisten(String message){SimpleDateFormat sdf =newSimpleDateFormat("yyyy-MM-dd HH:mm:ss");String dateStr = sdf.format(newDate());
        log.info("my-kafka-consumer-groupId-02 消息消费成功,当前时间:{},消息内容:{}", dateStr, message);}}

消费端二的 Consumer Group 我们使用 Spring EL 表达式，在每个 groupId 上配合 UUID 生成其后缀 UUID 来实现，保证了消费组不重复，来实现广播消费的效果。

Kafak 消息广播消费结果验证

我们触发消息发送得到如下结果：

2024-10-27 19:42:35.272  INFO 26972 ---[nio-8086-exec-2] c.o.s.kafka.producer.MyKafkaProducer     : 完成消息发送,当前时间:2024-10-27 16:42:35
2024-10-27 19: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 19:42:35.378  INFO 26972 ---[-consumer-0-C-1] c.o.s.kafka.consumer.MyKafkaConsumer     : 消息消费成功,当前时间:2024-10-27 16:42:35,消息内容:第一条 kafka 消息

可以看到一条消息被消费了两次，也就是两个消费端都消费到了消息，结果符合预期。

如有不正确的地方欢迎各位指出纠正。