Springboot详解RocketMQ实现广播消息流程

 更新时间:2022年06月22日 11:35:21   作者:12程序猿  
RocketMQ作为一款纯java、分布式、队列模型的开源消息中间件,支持事务消息、顺序消息、批量消息、定时消息、消息回溯等,本篇我们了解如何实现广播消息

RocketMQ消息模式主要有两种:广播模式、集群模式(负载均衡模式)

广播模式是每个消费者,都会消费消息;

负载均衡模式是每一个消费只会被某一个消费者消费一次;

我们业务上一般用的是负载均衡模式,当然一些特殊场景需要用到广播模式,比如发送一个信息到邮箱,手机,站内提示;

我们可以通过@RocketMQMessageListenermessageModel属性值来设置,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广播消息内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • java中调用https请求忽略ssl证书认证代码示例

    java中调用https请求忽略ssl证书认证代码示例

    在网络请求中经常会遇到需要忽略证书认证的情况,这篇文章主要介绍了java中调用https请求忽略ssl证书认证的相关资料,文中通过代码示例介绍的非常详细,需要的朋友可以参考下
    2024-10-10
  • Spring Boot使用GridFS实现文件的上传和下载方式

    Spring Boot使用GridFS实现文件的上传和下载方式

    这篇文章主要介绍了Spring Boot使用GridFS实现文件的上传和下载方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2021-10-10
  • Maven中的库repository详解

    Maven中的库repository详解

    Maven中要配置库,可以有多种方式,最直接的是在项目中的pom.xml文件中,通过<repositories>配置库,这样配置的库仅适用于当前项目,这篇文章主要介绍了Maven中的库(repository),需要的朋友可以参考下
    2024-01-01
  • Springboot MultipartFile文件上传与下载的实现示例

    Springboot MultipartFile文件上传与下载的实现示例

    在Spring Boot项目中,可以使用MultipartFile类来处理文件上传和下载操作,本文就详细介绍了如何使用,具有一定的参考价值,感兴趣的可以了解一下
    2023-08-08
  • Spring3 MVC请求参数获取的几种方法小结

    Spring3 MVC请求参数获取的几种方法小结

    本篇文章主要介绍了Spring3 MVC请求参数获取的几种方法小结,非常具有实用价值,需要的朋友可以参考下。
    2017-03-03
  • java笔记学习之操作符

    java笔记学习之操作符

    本篇文章给大家分享了关于Java中操作符的相关知识点以及难点总结,有需要的朋友参考学习下吧。
    2018-04-04
  • 基于controller使用map接收参数的注意事项

    基于controller使用map接收参数的注意事项

    这篇文章主要介绍了基于controller使用map接收参数的注意事项,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2021-10-10
  • 解决Springboot @Autowired 无法注入问题

    解决Springboot @Autowired 无法注入问题

    WebappApplication 一定要在包的最外层,否则Spring无法对所有的类进行托管,会造成@Autowired 无法注入。接下来给大家介绍解决Springboot @Autowired 无法注入问题,感兴趣的朋友一起看看吧
    2018-08-08
  • springboot各种格式转pdf的实例代码

    springboot各种格式转pdf的实例代码

    这篇文章主要介绍了springboot各种格式转pdf的实例代码,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2021-01-01
  • Spring AOP底层机制之代理模式

    Spring AOP底层机制之代理模式

    如何实现在不修改源码的基础上实现代码功能的增强呢?spring为我们提供了代理模式。所谓的代理模式通俗来说就是一个中介,它给某一个对象提供一个代理对象,并由代理对象控制原对象的引用,从而实现在不修改源码的基础上实现代码功能的增强
    2022-08-08

最新评论