Kafka 之消息广播消费

2024-11-02 19:20:38

前言：

上一篇我们分享了 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 即可，案例代码如下：

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 消息

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

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

码农公寓

Kafka 广播消息案例演示

相关文章