Springboot详解RocketMQ实现广播消息流程
更新时间：2022年06月22日 11:35:21 作者：12程序猿
RocketMQ作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等，本篇我们了解如何实现广播消息
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);
}
}- 重启服务，测试 广播模式消费
广播模式消费下，两个消费者都消费到Topic的所有消息。
测试成功！
到此这篇关于Springboot详解RocketMQ实现广播消息流程的文章就介绍到这了,更多相关Springboot广播消息内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
利用ClasserLoader实现jar包加载并调用里面的方法
classloader即是类加载,虚拟机把描述类的数据从class字节码文件加载到内存,并对数据进行检验、转换解析和初始化,了解java的类加载机制,可以快速解决运行时的各种加载问题并快速定位其背后的本质原因,本文介绍了如何利用ClasserLoader来实现jar包加载并调用里面的方法2024-09-09
下面小编就为大家分享一篇浅谈Java 将图片打包到jar中的路径问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2018-02-02
通过java中Controller层，来接受请求，数据库查询到的数据进行封装，然后使用ExcelUtils进行输出，接下来通过本文给大家分享Java实现导出Excel功能的实例代码，感兴趣的朋友跟随小编一起看看吧2021-11-11
Java 泛型（generics）是 Jdk 5 中引入的一个新特性， 泛型提供了编译时类型安全检测机制， 该机制允许程序员在编译时检测到非法的类型。本文将通过示例详解Java泛型的定义与使用，需要的可以参考一下2022-08-08
这篇文章主要介绍了MyBatis逆向工程基本操作及代码实例,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2020-07-07
这篇文章主要介绍了Mybatisplus自动填充实现方式及代码示例,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2020-11-11
这篇文章主要介绍了在Linux上搭建一个Java部署环境,安装jdk有很多种方式，但是我们这里推荐的是使用yum直接安装openjdk，本文给大家介绍的非常详细，需要的朋友可以参考下2023-01-01
Spring IoC 容器是一个管理 Bean 的容器,在 Spring 的定义中,它要求所有的 IoC 容器都需要实现接口 BeanFactory,它是一个顶级容器接口,这篇文章主要介绍了SpringBoot中的Bean装配详解,需要的朋友可以参考下2024-04-04
这篇文章主要为大家详细介绍了支付宝APP支付（IOS手机端+java后台）版，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2019-05-05
这篇文章主要介绍了java集合map取key使用示例,需要的朋友可以参考下2014-02-02
最新评论