技术活,该赏
关注+一键三连(点赞,评论,收藏)再看,养成好习惯
RocketMQ使用教程相关系列 目录
第一节:介绍
消费者采用广播的方式消费消息,每个消费者消费的消息都是相同的
限制:
不支持顺序消息
消费者消费模式
消费者的消费模式分为两种
- 负载均衡模式:消费者采用负载均衡方式消费消息,多个消费者共同消费队列消息,每个消费者处理的消息不同
- 广播模式:消费者采用广播的方式消费消息,每个消费者消费的消息都是相同的
第二节:广播消息-生产者和消息者步骤说明
广播消息生产者代码实现步骤
1.创建消息生产者producer,并制定生产者组名
2.指定Nameserver地址
3.启动producer
4.创建消息对象集合,指定主题Topic、Tag和消息体
5.发送集合消息
6.关闭生产者producer
注:与批量消息的生产者代码一模一样
广播消息消费者代码实现步骤
1.创建消费者Consumer,制定消费者组名
2.指定Nameserver地址
3.默认均衡轮询消费模式 改为广播模式
4.订阅主题Topic和Tag
5.设置回调函数,处理消息
6.启动消费者consumer
注意:消费者的 Topic 和 Tag 需要和生产者保持一致
第三节:广播消息生产者
public class Producer {
public static void main(String[] args) throws Exception {
// 1.创建消息生产者producer,并制定生产者组名
DefaultMQProducer producer = new DefaultMQProducer("demo_producer_broadcasting_group");
// 2.指定Nameserver地址
producer.setNamesrvAddr("192.168.88.131:9876");
// 3.启动producer
producer.start();
System.out.println("生产者启动");
List<Message> msgs = new ArrayList<Message>();
// 4.创建消息对象,指定主题Topic、Tag和消息体
/**
* 参数一:消息主题Topic
* 参数二:消息Tag
* 参数三:消息内容
*/
for (int i = 0; i < 20; i++) {
Message msg = new Message("Topic_broadcasting_demo", "Tag_broadcasting_demo",
("Hello 虚竹,这是广播消息" + i).getBytes());
msgs.add(msg);
}
// 5.发送消息
SendResult result = producer.send(msgs);
// 发送状态
SendStatus status = result.getSendStatus();
System.out.println("发送结果:" + result);
// 线程睡1秒
TimeUnit.SECONDS.sleep(1);
// 6.关闭生产者producer
producer.shutdown();
System.out.println("生产者关闭");
}
}
效果:
第四节:广播消息消费者A
public class ConsumerA {
public static void main(String[] args) throws Exception {
// 1.创建消费者Consumer,制定消费者组名
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo_producer_broadcasting_group");
// 2.指定Nameserver地址
consumer.setNamesrvAddr("192.168.88.131:9876");
// 默认均衡轮询消费模式 改为广播模式
consumer.setMessageModel(MessageModel.BROADCASTING);
// 3.订阅主题Topic和Tag
consumer.subscribe("Topic_broadcasting_demo", "*");
// 4.设置回调函数,处理消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
// 接受消息内容
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
System.out.println(
"A----consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
// 5.启动消费者consumer
consumer.start();
System.out.println("消费者启动");
}
}
效果:
第五节:广播消息消费者B
public class ConsumerB {
public static void main(String[] args) throws Exception {
// 1.创建消费者Consumer,制定消费者组名
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo_producer_broadcasting_group");
// 2.指定Nameserver地址
consumer.setNamesrvAddr("192.168.88.131:9876");
// 默认均衡轮询消费模式 改为广播模式
consumer.setMessageModel(MessageModel.BROADCASTING);
// 3.订阅主题Topic和Tag
consumer.subscribe("Topic_broadcasting_demo", "*");
// 4.设置回调函数,处理消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
// 接受消息内容
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
System.out.println(
"B----consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
// 5.启动消费者consumer
consumer.start();
System.out.println("消费者启动");
}
}
效果:
第六节:总结
消费模式默认为负载均衡模式
修改消费模式的方法是: consumer.setMessageModel();
设置为广播模式:consumer.setMessageModel(MessageModel.BROADCASTING);
设置负载均衡模式:consumer.setMessageModel(MessageModel.CLUSTERING);
参考:
顺序消息是否支持集群消费和广播消费?
https://help.aliyun.com/knowledge_detail/54357.html