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广播消息内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

springboot2.6.3读取不到nacos上的配置文件问题
这篇文章主要介绍了springboot2.6.3读取不到nacos上的配置文件问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2023-07-07
mybatis省略@Param注解操作
这篇文章主要介绍了mybatis省略@Param注解操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-11-11
Java加载ICC文件的方法和示例代码
ICC文件,通常用于颜色管理,定义了如何将一个颜色空间转换为另一个颜色空间,在Java中,我们可能需要加载这些文件来进行颜色转换或管理,本文将为您提供加载ICC文件的方法和示例代码,需要的朋友参考下吧
2023-08-08
Base64加解密的实现方式实例详解
这篇文章主要介绍了Base64加解密的实现方式实例详解的相关资料,这里提供了实现实例，帮助大家学习理解这部分内容，需要的朋友可以参考下
2017-08-08
Spring Boot异步线程间数据传递的四种方式
这篇文章主要为大家介绍了Spring Boot异步线程间数据传递的四种方式详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2023-01-01
浅析Spring的事务实现原理
这篇文章主要为大家详细介绍了Spring中事务实现的原理，文中的示例代码讲解详细，对我们学习Spring有一定的帮助，需要的可以参考一下
2022-11-11
带你了解Java数据结构和算法之哈希表
这篇文章主要为大家介绍了Java数据结构和算法之哈希表，具有一定的参考价值，感兴趣的小伙伴们可以参考一下，希望能够给你带来帮助
2022-01-01
Java图形化界面编程实现简单计算器
这篇文章主要介绍了Java图形化界面编程实现简单计算器，下面文章围绕Java图形化界面编程实现简单计算器的相关资料展开详细内容，具有一定的参考价值，需要的小伙伴可以参考一下
2022-01-01
java 直接调用python脚本，并传递参数代码实例
这篇文章主要介绍了java调用python脚本传递参数的方法，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-04-04
springboot整合activity自动部署及部署文件命名流程
这篇文章主要介绍了springboot整合activity自动部署及部署文件命名流程，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-09-09