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

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

时间: 2022-06-21

springboot整合rocketmq实现分布式事务

1 执行流程 (1) 发送方向 MQ 服务端发送消息. (2) MQ Server 将消息持久化成功之后,向发送方 ACK 确认消息已经发送成功,此时消息为半消息. (3) 发送方开始执行本地事务逻辑. (4) 发送方根据本地事务执行结果向 MQ Server 提交二次确认(Commit 或是 Rollback),MQ Server 收到Commit 状态则将半消息标记为可投递,订阅方最终将收到该消息:MQ Server 收到 Rollback 状态则删除半消息,订阅方将不会接受该消息. (5)

浅谈Springboot整合RocketMQ使用心得

一.阿里云官网---帮助文档 https://help.aliyun.com/document_detail/29536.html?spm=5176.doc29535.6.555.WWTIUh 按照官网步骤,创建Topic.申请发布(生产者).申请订阅(消费者) 二.代码 1.配置: public class MqConfig { /** * 启动测试之前请替换如下 XXX 为您的配置 */ public static final String PUBLIC_TOPIC = "test"

Springboot RocketMq实现过程详解

首先,在虚拟机上安装rocketmq和rocketMq可视化控制,安装不做描述. 1.pom.xml文件添加依赖 mq的版本与连接的rocketmq版本保持一致 <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-remoting</artifactId> <version>4.4.0</version> </depende

SpringBoot整合RocketMQ实现消息发送和接收的详细步骤

我们使用主流的SpringBoot框架整合RocketMQ来讲解,使用方便快捷: 最终项目结构如下: 具体步骤如下: 第一步:新建SpringBoot项目rocketmq-test,引入rocketmq依赖,以及项目配置 <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <vers

springBoot整合RocketMQ及坑的示例代码

版本: JDK:1.8 springBoot:1.5.10 rocketMQ:4.2.0 pom 配置: <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.5.10.RELEASE</version> </parent> <d

解决SpringBoot整合RocketMQ遇到的坑

应用场景 在实现RocketMQ消费时,一般会用到@RocketMQMessageListener注解定义Group.Topic以及selectorExpression(数据过滤.选择的规则)为了能支持动态筛选数据,一般都会使用表达式,然后通过apollo或者cloud config进行动态切换. 引入依赖 <!-- RocketMq Spring Boot Starter--> <dependency> <groupId>org.apache.rocketmq<

RocketMQ整合SpringBoot实现生产级二次封装

目录 前言说明 一.为什么要二次封装 1.1 二次封装不同观点 1.2 封装的抽离点 1.3 设计模式的应用 二.二次封装核心要点 2.1 二次封装核心点 2.1.1 封装主要讨论点 2.1.2 发送/消费的几种消息实体 2.2 RocketMQTemplate封装 2.2.1 封装基础实体类 2.2.2 RocketMQTemplate 3.2.3 增强RocketMQTemplate 2.3 RocketMQListener封装 2.4 广播消息的应用场景 2.3 代码封装完结测试 前言说明

SpringBoot中使用RocketMQ的示例代码

目录 1 订单微服务发送消息 1.1 订单微服务添加rocketmq的依赖 1.2 添加配置 1.3 编写测试代码 1.4 测试 2 用户微服务订阅消息 2.1 用户微服务增加rocketmq依赖 2.2 修改主类,启动nacos客户端 2.3 修改配置文件 2.4 编写消息接收服务 2.5 测试 接下来我们模拟一种场景:商品下单成功之后,向下单用户发送短信.以此来示例SpringBoot中RocketMQ的使用方式. 1 订单微服务发送消息 1.1 订单微服务添加rocketmq的依赖 <!-

Springboot&nbsp;整合&nbsp;RocketMQ&nbsp;收发消息

Springboot 整合 RocketMQ 收发消息 创建springboot项目 pom.xml添加rocketmq-spring-boot-starter依赖. <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <version>2.1.0</version>

SpringBoot+RabbitMQ方式收发消息的实现示例

本篇会和SpringBoot做整合,采用自动配置的方式进行开发,我们只需要声明RabbitMQ地址就可以了,关于各种创建连接关闭连接的事都由Spring帮我们了~ 交给Spring帮我们管理连接可以让我们专注于业务逻辑,就像声明式事务一样易用,方便又高效. 祝有好收获,先赞后看,快乐无限. 本文代码:   https://gitee.com/he-erduo/spring-boot-learning-demo https://github.com/he-erduo/spring-boot-lea

SpringBoot 整合Redis 数据库的方法

Redis简介 Redis(官网: https://redis.io )是一个基于内存的日志型可持久化的缓存数据库,保存形式为key-value格式,Redis完全免费开源,它使用ANSI C语言编写.与其他的key - value缓存产品一样,Redis具有以下三个特点. • Redis支持数据的持久化,可以将内存中的数据保存在磁盘中,重启的时候可以再次加载进行使用: • Redis不仅支持简单的key-value类型数据,同时还提供字符串.链表.集合.有序集合和哈希等数据结构的存储: • R

spring整合JMS实现同步收发消息(基于ActiveMQ的实现)

本文介绍了spring整合JMS实现同步收发消息(基于ActiveMQ的实现),分享给大家,具体如下: 1. 安装ActiveMQ 注意:JDK版本需要1.7及以上才行 到Apache官方网站下载最新的ActiveMQ的安装包,并解压到本地目录下,下载链接如下:http://activemq.apache.org/download.html,解压后的目录结构如下: bin目录结构如下: 如果我们是32位的机器,就双击win32目录下的activemq.bat,如果是64位机器,则双击win64目

SpringBoot整合Redis、ApachSolr和SpringSession的示例

本文介绍了SpringBoot整合Redis.ApachSolr和SpringSession,分享给大家,具体如下: 一.简介 SpringBoot自从问世以来,以其方便的配置受到了广大开发者的青睐.它提供了各种starter简化很多繁琐的配置.SpringBoot整合Druid.Mybatis已经司空见惯,在这里就不详细介绍了.今天我们要介绍的是使用SpringBoot整合Redis.ApacheSolr和SpringSession. 二.SpringBoot整合Redis Redis是大家比

Springboot整合log4j2日志全解总结

在项目推进中,如果说第一件事是搭Spring框架的话,那么第二件事情就是在Sring基础上搭建日志框架,我想很多人都知道日志对于一个项目的重要性,尤其是线上Web项目,因为日志可能是我们了解应用如何执行的唯一方式. 在18年大环境下,更多的企业使用Springboot和Springcloud来搭建他们的企业微服务项目 ,此篇文章是博主在实践中用Springboot整合log4j2日志的总结. 常用日志框架 java.util.logging:是JDK在1.4版本中引入的Java原生日志框架 Lo

SpringBoot 整合Jest实例代码讲解

[1]添加Elasticsearch-starter pom文件添加starter如下: <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-elasticsearch</artifactId> </dependency> SpringBoot默认支持两种技术和Elasticsearch进行交互:Sp

SpringBoot整合FastJson过程解析

这篇文章主要介绍了SpringBoot整合FastJson过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 一.Maven依赖 <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.33</version> </dependency&

springboot整合apache ftpserver详细教程(推荐)

一.Apache ftpserver相关简介 Apache FtpServer是100%纯Java FTP服务器.它被设计为基于当前可用的开放协议的完整且可移植的FTP服务器引擎解决方案.FtpServer可以作为Windows服务或Unix / Linux守护程序独立运行,也可以嵌入Java应用程序中.我们还提供对Spring应用程序内集成的支持,并以OSGi捆绑软件的形式提供我们的发行版.默认的网络支持基于高性能异步IO库Apache MINA.使用MINA,FtpServer可以扩展到大量