RocketMQ消息发送之广播模式

创作时间:

作者:

@小白创作中心

RocketMQ消息发送之广播模式

引用

CSDN

https://m.blog.csdn.net/qq_53847859/article/details/140938543

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");