RocketMQ消息发送之广播模式

news2024/12/25 9:02:48

前言

在前面的文章中我们回顾了RocketMQ的顺序消息和乱序消息，以及里面包含的乱序消息和全局消息，RocketMQ支持多种消息类型和消费模式

今天这篇文章主要介绍RocketMQ的广播消息。希望文章能为正在学习RocketMQ相关知识的大佬们提供帮助！

在这里插入图片描述

广播模式概述

在RocketMQ中，广播消息是一种特殊的消息消费模式。在广播模式下，每条消息都会被推送到集群内所有注册过的客户端，保证消息至少被每台机器消费一次。与集群模式不同，集群模式下，同一个消费者组内的消息只会被消费一次，而广播模式下，每个消费者都会收到并消费相同的消息。

广播模式特点

消息重复消费：广播模式下，每条消息都会被推送到所有订阅了Topic的消费者，这意味着消息可能会被重复消费多次。
不支持顺序消息：广播模式下不支持顺序消息，因为顺序消息需要保证消息的消费顺序，这在广播模式下很难实现的，毕竟所有消费者都会同时收到消息。
不支持重置消费位点：广播模式下，消费进度在客户端维护，不支持在服务端重置消费位点。如果客户端重启，它将从最新消息开始消费，并不会继续消费之前未处理完的消息。
消费失败不重试：在广播模式下，如果消息消费失败，RocketMQ不会自动重试。因此，我们很多时候需要自行处理消费失败的情况，比如记录日志、发送告警或这手动重试。

广播模式应用场景

当需要将一条消息通知到多个系统或组件时，可以使用广播消息模式。比如，在某多多或某宝这些电商系统中，订单状态变更需要通知库存系统、物流系统等多个下游系统，这时候就可以使用广播消息。

除此之外，在日志收集系统中，我们可以将日志消息以广播模式发送到多个日志处理组件，每个组件根据自身的职责进行日志处理和分析。

还有，在实时通知推送场景中，比如发送短信、邮件、APP推送等等，我们可以使用广播消息模式将通知消息推送给所有相关的接收者。

手动实现

首先还是使用以往的方式启动搭建的RocketMQ集群。

消息生产者比较简单，代码如下：

        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        producer.start();
        for (int i = 0; i < 100; i++){
            Message msg = new Message("TopicTest",
                    "TagA",
                    "xiaowei",
                    ("Hello world"+i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        producer.shutdown();

在RocketMQ中，实现广播消息模式非常简单。我们可以通过设置消息模型为MessageModel.BROADCASTING来启用广播模式。
接下来是消费者代码，如下（流程在注释中）：

// 创建DefaultMQPushConsumer实例 
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");  
// 消费者从队列的开始位置消费消息  
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);  
// 消费者的消息模型为广播模式，消息会被广播给所有消费者  
consumer.setMessageModel(MessageModel.BROADCASTING);  
// 订阅指定的Topic，使用"*"订阅Topic下的所有消息  
consumer.subscribe("TopicTest", "*");  
// 注册消息监听器
consumer.registerMessageListener(new MessageListenerConcurrently() {  
    @Override  
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,  
                                                    ConsumeConcurrentlyContext context) {  
        // 遍历消息列表，处理每一条消息  
        for(MessageExt msg:msgs){  
            // 打印消息内容  
            System.out.println("消息内容："+new String(msg.getBody()));  
        }  
        // 返回消费状态，消息已成功消费  
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
    }  
});  
// 启动消费者实例  
consumer.start();  
// 打印启动成功的提示信息  
System.out.printf("Broadcast Consumer Started.%n");