RocketMQ实践之广播模式（四）

- 日理万妓 2022-09-15 12:27 208阅读 0赞

**发送消息：**

--------------------

package my.demo.broad;
    
    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;
    
    public class BroadcastProducer { 
        public static void main(String[] args) throws Exception { 
            DefaultMQProducer producer = new DefaultMQProducer("broad_group");
            producer.setNamesrvAddr("localhost:9876");
            producer.start();
    
            for (int i = 0; i < 100; i++){ 
                Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            }
            producer.shutdown();
        }
    }

**接收消息：**

--------------------

package my.demo.broad;
    
    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;
    
    public class BroadcastConsumer { 
        public static void main(String[] args) throws Exception { 
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("broad_group");
            consumer.setNamesrvAddr("localhost:9876");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            //set to broadcast mode
            consumer.setMessageModel(MessageModel.BROADCASTING);
    
            consumer.subscribe("TopicTest", "TagA");
    
            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");
        }
    }