一、概述

参考链接
http://rocketmq.apache.org/docs/broadcast-example/
广播，区别于默认的集群模式，在集群模式下，一条消息，同一个Group+tag的Consumer集群中只有一个Consumer能够消费，而广播模式，所有的消费者都会收到。

二、BroadcastConsumer 广播消费者

需要设置 consumer.setMessageModel(MessageModel.BROADCASTING);,其他的和一般的消费者没有什么区别
顺道说一下 ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET，这个类似于Kafka的offset，可以设定读取历史消息，如果已经读取过，可以更改下groupName，然后重新读取，具体的原理待后续完成OFFSET的DEMO后完善

package com.asd.rocket.controller.brod;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import java.util.List;

/**
 * @author zhangluping@sinosoft.com.cn
 * @date 2019/4/11 14:32
 */
public class BroadcastConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name_N2");
        consumer.setNamesrvAddr("10.1.11.155:9876");
        // 设置起始的读取位置
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //set to broadcast mode
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.subscribe("qqq", "TagA || TagB || TagC");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Broadcast Consumer Started.%n");
    }
}

三、Producer生产者

与一般生产者无异

package com.asd.rocket.controller.test;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

/**
 * @author zhangluping@sinosoft.com.cn
 * @date 2019/4/10 10:08
 */
public class Producer {
    public static void main(String[] args) throws Exception{
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new
                DefaultMQProducer("please_rename_unique_group_name");
        // Specify name server addresses.
        producer.setNamesrvAddr("10.1.11.155:9876");
        producer.setSendMsgTimeout(13000);
        producer.setCreateTopicKey("AUTO_CREATE_TOPIC_KEY");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 1; i++) {
            /*
             * Create a message instance, specifying topic, tag and message body.
             * 三个参数
             * Topic、Tag、MessageBody
             */
            Message msg = new Message("qqq","TagA" , "keu1",("Hello RocketMQ " +i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
        System.out.println(1);
    }
}