软件开发项目经验_网页上的视频怎么下载_经典seo伪原创_营销网络图

注意事项

集群消费模式下，每⼀条消息都只会被分发到⼀台机器上处理。如果需要被集群下的每⼀台机器都处理，请使⽤⼴播模式。
集群消费模式下，不保证每⼀次失败重投的消息路由到同⼀台机器上。

广播消息：消息被每个消费者都消费
集群消息：消息只能被一个消费者消费（默认）

⼴播消费模式下不⽀持顺序消息。
⼴播消费模式下不⽀持重置消费位点。
每条消息都需要被相同订阅逻辑的多台机器处理。
⼴播模式下服务端不维护消费进度，消费进度在客户端维护
⼴播模式下，消息队列 RocketMQ 版保证每条消息⾄少被每台客户端消费⼀次，但是并不会重投
消费失败的消息，因此业务⽅需要关注消费失败的情况。
⼴播模式下，客户端每⼀次重启都会从最新消息消费。客户端在被停⽌期间发送⾄服务端的消息将会被⾃动跳过，请谨慎选择。
⼴播模式下，每条消息都会被⼤量的客户端重复处理，因此推荐尽可能使⽤集群模式。

BroadcastProducer.java

package com.example.rocketmq.demo.broadcasting;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("ProducerGroupName");// Specify name server addresses.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();}
}

BroadcastConsumer.java

package com.example.rocketmq.demo.broadcasting;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("group2");// Specify name server addresses.consumer.setNamesrvAddr("localhost:9876");consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.setMessageModel(MessageModel.BROADCASTING); //设置广播模式consumer.subscribe("TopicTest", "TagA || TagC || TagD");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic 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");}
}