title: 广播消息示例 date: 2017/12/21

categories: 文档翻译

Broadcasting

What is broadcasting

Broadcasting is sending a message to all subscribers of a topic. If you want all subscribers receive messages about a topic, broadcasting is a good choice.

广播

什么是广播

广播就是向一个主题的所有订阅者发送同一条消息。如果你想让一个主题的所有订阅者收到消息，广播是一个很好的选择。

Producer example

  1. public class BroadcastProducer {
  2.     public static void main(String[] args) throws Exception {
  3.         DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
  4.         producer.start();
  6.         for (int i = 0; i < 100; i++){
  7.             Message msg = new Message("TopicTest",
  8.                 "TagA",
  9.                 "OrderID188",
  10.                 "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
  11.             SendResult sendResult = producer.send(msg);
  12.             System.out.printf("%s%n", sendResult);
  13.         }
  14.         producer.shutdown();
  15.     }
  16. }

生产者示例

  1. public class BroadcastProducer {
  2.     public static void main(String[] args) throws Exception {
  3.         DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
  4.         producer.start();
  6.         for (int i = 0; i < 100; i++){
  7.             Message msg = new Message("TopicTest",
  8.                 "TagA",
  9.                 "OrderID188",
  10.                 "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
  11.             SendResult sendResult = producer.send(msg);
  12.             System.out.printf("%s%n", sendResult);
  13.         }
  14.         producer.shutdown();
  15.     }
  16. }

Consumer example

  1. public class BroadcastConsumer {
  2.     public static void main(String[] args) throws Exception {
  3.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");
  5.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  7.         //set to broadcast mode
  8.         consumer.setMessageModel(MessageModel.BROADCASTING);
  10.         consumer.subscribe("TopicTest", "TagA || TagC || TagD");
  12.         consumer.registerMessageListener(new MessageListenerConcurrently() {
  14.             @Override
  15.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
  16.                 ConsumeConcurrentlyContext context) {
  17.                 System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
  18.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  19.             }
  20.         });
  22.         consumer.start();
  23.         System.out.printf("Broadcast Consumer Started.%n");
  24.     }
  25. }

消费者示例

  1. public class BroadcastConsumer {
  2.     public static void main(String[] args) throws Exception {
  3.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");
  5.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  7.         //set to broadcast mode
  8.         consumer.setMessageModel(MessageModel.BROADCASTING);
  10.         consumer.subscribe("TopicTest", "TagA || TagC || TagD");
  12.         consumer.registerMessageListener(new MessageListenerConcurrently() {
  14.             @Override
  15.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
  16.                 ConsumeConcurrentlyContext context) {
  17.                 System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
  18.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  19.             }
  20.         });
  22.         consumer.start();
  23.         System.out.printf("Broadcast Consumer Started.%n");
  24.     }
  25. }