rocketmq消息模式主要有两种：广播模式、集群模式（负载均衡模式）
广播模式是每个消费者，都会消费消息；
负载均衡模式是每一个消费只会被某一个消费者消费一次；
我们业务上一般用的是负载均衡模式，当然一些特殊场景需要用到广播模式，比如发送一个信息到邮箱，手机，站内提示；
我们可以通过@RocketMQMessageListener的messageModel属性值来设置，MessageModel.BROADCASTING是广播模式，MessageModel.CLUSTERING是默认集群负载均衡模式
下面来介绍下 springboot+rockermq 整合实现 广播消息
创建Springboot项目，添加rockermq 依赖
org.apache.rocketmq rocketmq-spring-boot-starter 2.2.1
配置rocketmq
# 端口server: port: 8083# 配置 rocketmqrocketmq: name-server: 127.0.0.1:9876 #生产者 producer: #生产者组名，规定在一个应用里面必须唯一 group: group1 #消息发送的超时时间 默认3000ms send-message-timeout: 3000 #消息达到4096字节的时候，消息就会被压缩。默认 4096 compress-message-body-threshold: 4096 #最大的消息限制，默认为128K max-message-size: 4194304 #同步消息发送失败重试次数 retry-times-when-send-failed: 3 #在内部发送失败时是否重试其他代理，这个参数在有多个broker时才生效 retry-next-server: true #异步消息发送失败重试的次数 retry-times-when-send-async-failed: 3
生产端：新建一个 controller 来做消息发送
生产端按正常发送逻辑发送消息即可
package com.example.springbootrocketdemo.controller;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* 广播消息
* @author qzz
*/
@RestController
public class RocketMQBroadCOntroller {
@Autowired
private RocketMQTemplate rocketMQTemplate;
/**
* 发送广播消息
*/
@RequestMapping("/testBroadSend")
public void testSyncSend(){
//参数一：topic 如果想添加tag,可以使用"topic:tag"的写法
//参数二：消息内容
for(int i=0;i<10;i++){
rocketMQTemplate.convertAndSend("test-topic-broad","test-message"+i);
}
}
}创建两个消费者来消费消息
我们先集群负载均衡测试，加上messageModel=MessageModel.CLUSTERING
消费者1：
package com.example.springbootrocketdemo.config; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Service; /** * 广播消息 * 配置RocketMQ监听 * MessageModel.CLUSTERING:集群模式 * MessageModel.BROADCASTING:广播模式 * @author qzz */ @Service @RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.CLUSTERING) public class RocketMQBroadConsumerListener implements RocketMQListener{ @Override public void onMessage(String s) { System.out.println("集群模式 消费者1，消费消息："+s); } }
消费者2： 与消费者1在 同一个consumerGroup 和 topic
package com.example.springbootrocketdemo.config; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Service; /** * 广播消息 * 配置RocketMQ监听 * MessageModel.CLUSTERING:集群模式 * MessageModel.BROADCASTING:广播模式 * @author qzz */ @Service @RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.CLUSTERING) public class RocketMQBroadConsumerListener2 implements RocketMQListener{ @Override public void onMessage(String s) { System.out.println("集群模式 消费者2，消费消息："+s); } }
启动服务，测试 集群模式消费
集群模式测试： 两个消费者平摊 消息
把上面两个消费者的 messageModel 属性值修改成 广播模式
消费者1：
package com.example.springbootrocketdemo.config; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Service; /** * 广播消息 * 配置RocketMQ监听 * MessageModel.CLUSTERING:集群模式 * MessageModel.BROADCASTING:广播模式 * @author qzz */ @Service @RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.BROADCASTING) public class RocketMQBroadConsumerListener implements RocketMQListener{ @Override public void onMessage(String s) { System.out.println("广播消息1 广播模式，消费消息："+s); } }
消费者2： 与消费者1在 同一个consumerGroup 和 topic
package com.example.springbootrocketdemo.config; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Service; /** * 广播消息 * 配置RocketMQ监听 * MessageModel.CLUSTERING:集群模式 * MessageModel.BROADCASTING:广播模式 * @author qzz */ @Service @RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.BROADCASTING) public class RocketMQBroadConsumerListener2 implements RocketMQListener{ @Override public void onMessage(String s) { System.out.println("广播消息2 广播模式，消费消息："+s); } }
重启服务，测试 广播模式消费
