这篇文章主要介绍“Springboot中RocketMQ怎么实现广播消息”,在日常操作中,相信很多人在Springboot中RocketMQ怎么实现广播消息问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”Springboot中RocketMQ怎么实现广播消息”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!
RocketMQ消息模式主要有两种:广播模式、集群模式(负载均衡模式)
广播模式是每个消费者,都会消费消息;
负载均衡模式是每一个消费只会被某一个消费者消费一次;
我们业务上一般用的是负载均衡模式,当然一些特殊场景需要用到广播模式,比如发送一个信息到邮箱,手机,站内提示;
我们可以通过@RocketMQMessageListener
的messageModel
属性值来设置,MessageModel.BROADCASTING
是广播模式,MessageModel.CLUSTERING
是默认集群负载均衡模式
下面来介绍下 springboot+rockermq 整合实现 广播消息
创建Springboot项目,添加rockermq 依赖
<!--rocketMq依赖-->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.1</version>
</dependency>
配置rocketmq
# 端口
server:
port: 8083# 配置 rocketmq
rocketmq:
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<String> {
@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<String> {
@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<String> {
@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<String> {
@Override
public void onMessage(String s) {
System.out.println("广播消息2 广播模式,消费消息:"+s);
}
}
重启服务,测试 广播模式消费
到此,关于“Springboot中RocketMQ怎么实现广播消息”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注亿速云网站,小编会继续努力为大家带来更多实用的文章!
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。