20241130 RocketMQ本机安装与SpringBoot整合

一、RocketMQ简介

???1.1、核心概念

???1.2、应用场景

???1.3、架构设计

2、RocketMQ Server安装

3、RocketMQ可视化控制台安装与使用

4、SpringBoot整合RocketMQ实现消息发送和接收?

? ? ? ? 4.1、添加maven依赖

???4.2、yaml配置

???4.3、生产者

???4.4、消费者

???4.5、接口

???4.6、接口测试

一、RocketMQ简介

RocketMQ是由阿里捐赠给Apache的一款低延迟、高并发、高可用、高可靠的分布式消息中间件。经历了淘宝双十一的洗礼。RocketMQ既可为分布式应用系统提供异步解耦和削峰填谷的能力，同时也具备互联网应用所需的海量消息堆积、高吞吐、可靠重试等特性。

官方文档：https://rocketmq.apache.org/docs/quick-start/

github中文主页：https://github.com/apache/rocketmq/tree/master/docs/cn

1.1、核心概念

Topic：消息主题，一级消息类型，生产者向其发送消息。Message：生产者向Topic发送并最终传送给消费者的数据消息的载体。
消息属性：生产者可以为消息定义的属性，包含Message Key和Tag。
Message Key：消息的业务标识，由消息生产者（Producer）设置，唯一标识某个业务逻辑。
Message ID：消息的全局唯一标识，由消息队列RocketMQ系统自动生成，唯一标识某条消息。
Tag：消息标签，二级消息类型，用来进一步区分某个Topic下的消息分类
Producer：也称为消息发布者，负责生产并发送消息至Topic。
Consumer：也称为消息订阅者，负责从Topic接收并消费消息。
分区：即Topic Partition，物理上的概念。每个Topic包含一个或多个分区。
消费位点：每个Topic会有多个分区，每个分区会统计当前消息的总条数，这个称为最大位点MaxOffset；分区的起始位置对应的位置叫做起始位点MinOffset。
Group：一类生产者或消费者，这类生产者或消费者通常生产或消费同一类消息，且消息发布或订阅的逻辑一致。
Group ID：Group的标识。
队列：单个Topic下会由一到多个队列来存储消息。
Exactly-Once****投递语义：Exactly-Once投递语义是指发送到消息系统的消息只能被Consumer处理且仅处理一次，即使Producer重试消息发送导致某消息重复投递，该消息在Consumer也只被消费一次。
集群消费：一个Group ID所标识的所有Consumer平均分摊消费消息。例如某个Topic有9条消息，一个Group ID有3个Consumer实例，那么在集群消费模式下每个实例平均分摊，只消费其中的3条消息。
广播消费：一个Group ID所标识的所有Consumer都会各自消费某条消息一次。例如某个Topic有9条消息，一个Group ID有3个Consumer实例，那么在广播消费模式下每个实例都会各自消费9条消息。
定时消息：Producer将消息发送到消息队列RocketMQ服务端，但并不期望这条消息立马投递，而是推迟到在当前时间点之后的某一个时间投递到Consumer进行消费，该消息即定时消息。
延时消息：Producer将消息发送到消息队列RocketMQ服务端，但并不期望这条消息立马投递，而是延迟一定时间后才投递到Consumer进行消费，该消息即延时消息。
事务消息：RocketMQ提供类似X/Open XA的分布事务功能，通过消息队列RocketMQ的事务消息能达到分布式事务的最终一致。
顺序消息：RocketMQ提供的一种按照顺序进行发布和消费的消息类型，分为全局顺序消息和分区顺序消息。
全局顺序消息：对于指定的一个Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费。
分区顺序消息：对于指定的一个Topic，所有消息根据Sharding Key进行区块分区。同一个分区内的消息按照严格的FIFO顺序进行发布和消费。Sharding Key是顺序消息中用来区分不同分区的关键字段，和普通消息的Message Key是完全不同的概念。
消息堆积：Producer已经将消息发送到消息队列RocketMQ的服务端，但由于Consumer消费能力有限，未能在短时间内将所有消息正确消费掉，此时在消息队列RocketMQ的服务端保存着未被消费的消息，该状态即消息堆积。
消息过滤：Consumer可以根据消息标签（Tag）对消息进行过滤，确保Consumer最终只接收被过滤后的消息类型。消息过滤在消息队列RocketMQ的服务端完成。
消息轨迹：在一条消息从Producer发出到Consumer消费处理过程中，由各个相关节点的时间、地点等数据汇聚而成的完整链路信息。通过消息轨迹，您能清晰定位消息从Producer发出，经由消息队列RocketMQ服务端，投递给Consumer的完整链路，方便定位排查问题。
重置消费位点：以时间轴为坐标，在消息持久化存储的时间范围内（默认3天），重新设置Consumer对已订阅的Topic的消费进度，设置完成后Consumer将接收设定时间点之后由Producer发送到消息队列RocketMQ服务端的消息。
死信队列：死信队列用于处理无法被正常消费的消息。当一条消息初次消费失败，消息队列RocketMQ会自动进行消息重试；达到最大重试次数后，若消费依然失败，则表明Consumer在正常情况下无法正确地消费该消息。此时，消息队列RocketMQ不会立刻将消息丢弃，而是将这条消息发送到该Consumer对应的特殊队列中。消息队RocketMQ将这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），将存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。

1.2、应用场景

削峰填谷：诸如秒杀、抢红包、企业开门红等大型活动时皆会带来较高的流量脉冲，或因没做相应的保护而导致系统超负荷甚至崩溃，或因限制太过导致请求大量失败而影响用户体验，消息队列RocketMQ可提供削峰填谷的服务来解决该问题。
异步解耦：交易系统作为淘宝和天猫主站最核心的系统，每笔交易订单数据的产生会引起几百个下游业务系统的关注，包括物流、购物车、积分、流计算分析等等，整体业务系统庞大而且复杂，消息队列RocketMQ可实现异步通信和应用解耦，确保主站业务的连续性。
顺序收发：细数日常中需要保证顺序的应用场景非常多，例如证券交易过程时间优先原则，交易系统中的订单创建、支付、退款等流程，航班中的旅客登机消息处理等等。与先进先出FIFO（First In First Out）原理类似，消息队列RocketMQ提供的顺序消息即保证消息FIFO。
分布式事务一致性：交易系统、支付红包等场景需要确保数据的最终一致性，大量引入消息队列RocketMQ的分布式事务，既可以实现系统之间的解耦，又可以保证最终的数据一致性。
大数据分析：数据在“流动”中产生价值，传统数据分析大多是基于批量计算模型，而无法做到实时的数据分析，利用阿里云消息队列RocketMQ与流式计算引擎相结合，可以很方便的实现业务数据的实时分析。
分布式缓存同步：天猫双11大促，各个分会场琳琅满目的商品需要实时感知价格变化，大量并发访问数据库导致会场页面响应时间长，集中式缓存因带宽瓶颈，限制了商品变更的访问流量，通过消息队列RocketMQ构建分布式缓存，实时通知商品数据的变化。

1.3、架构设计

RocketMQ架构上主要分为四部分，如上图所示:

**Producer：**消息发布的角色，支持分布式集群方式部署。 Producer 通过 MQ 的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟。

**Consumer：**消息消费的角色，支持分布式集群方式部署。支持以 push 推， pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求。

NameServer： NameServer 是一个非常简单的 Topic 路由注册中心，其角色类似 Dubbo 中的zookeeper，支持 Broker 的动态注册与发现。主要包括两个功能： Broker 管理， NameServer 接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活；路由信息管理，每个 NameServer 将保存关于 Broker 集群的整个路由信息和用于客户端查询的队列信息。然后Producer 和 Conumser 通过 NameServer 就可以知道整个 Broker 集群的路由信息，从而进行消息的投递和消费。NameServer 通常也是集群的方式部署，各实例间相互不进行信息通讯。Broker 是向每一台 NameServer 注册自己的路由信息，所以每一个NameServer实例上面都保存一份完整的路由信息。当某个 NameServer 因某种原因下线了，Broker仍然可以向其它 NameServer 同步其路由信息， Producer,Consumer仍然可以动态感知

Broker的路由的信息。BrokerServer： Broker 主要负责消息的存储、投递和查询以及服务高可用保证，为了实现这些功能，Broker包含了以下几个重要子模块。

**1. Remoting Module：**整个Broker 的实体，负责处理来自 clients端的请求。

**2. Client Manager：**负责管理客户端(Producer/Consumer) 和维护 Consumer 的 Topic订阅信息

**3. Store Service：**提供方便简单的API接口处理消息存储到物理硬盘和查询功能。

4. HA Service： 高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能。

5. Index Service： 根据特定的Message key 对投递到 Broker的消息进行索引服务，以提供消息的快速查询。

2、RocketMQ Server安装

RocketMQ依赖Java环境，要求有JDK 1.8以上版本；

支持Windows和Linux平台；支持源码方式安装和使用已经编译好的安装包安装；

我们用windows平台安装RocketMQ Server编译好的安装包，来讲解RocketMQ；

下载地址：https://rocketmq.apache.org/dowloading/releases/

我们选择4.9.0二进制版本；

解压后的目录：

benchmark：里面是测试Demo；

bin：可执行脚本；conf：配置文件；

lib：依赖的jar包；

我们把rocketmq解压包放到D盘soft目录下，重命名 rocketmq ；

第一步：系统环境变量加两个配置

ROCKETMQ_HOME=“D:soft ocketmq”

NAMESRV_ADDR=“localhost:9876”

第二步：启动 Name Server

进入命令行执行：

.inmqnamesrv.cmd

第三步：启动 Broker

进入命令行执行：

.inmqbroker.cmd -n localhost:9876 autoCreateTopicEnable=true

第四步：发送和接收消息测试

进入命令行消息发送执行：

.in ools.cmd org.apache.rocketmq.example.quickstart.Producer

消息发送成功；

进入命令行消息接收执行：

.in ools.cmd org.apache.rocketmq.example.quickstart.Consumer

消息接收成功；

第五步：关闭服务

windows 下直接关闭命令行窗口即可；

3、RocketMQ可视化控制台安装与使用

RocketMQ 提供了一些扩展项目支持，地址： https://github.com/apache/rocketmq-externals

其中一个 rocketmq - connect - console 项目，就是我们需要的可视化控制台；

我们把整个项目下载下来，打开 rocketmq - console 项目；项目是 SpringBoot 开发；

打开 application.properties 配置文件，我们至少需要修改两个配置项；

server.port=8080 ，这个是可视化项目启动端口，我们改成 19876 ；

rocketmq.config.namesrvAddr= ，这个是指定 nameServer 地址和端口，我们暂时先搞成localhost:9876，等后面搞集群的话，要再修改；

修改后保存，在项目目录下进入命令行，执行：

D:soft ocketmqClient ocketmq-externals-master ocketmq-console

mvn clean package -Dmaven.test.skip=true

打包执行完后，在 target 目录，会生成一个可运行 jar rocketmq - console - ng - 2.0.0.jar

我们运行这个 jar ，进入命令行执行：

java -jar rocketmq-console-ng-2.0.0.jar

启动成功后，浏览器输入：http://localhost:19876/

说明一切OK；

4、SpringBoot整合RocketMQ实现消息发送和接收

4.1、添加maven依赖

org.apache.rocketmq rocketmq-spring-boot-starter 2.2.3

4.2、yaml配置

rocketmq:
name-server: localhost:9876
producer:
# 发送同一类消息的设置为同一个group，保证唯一
group: producer-group
# 发送消息超时时间，默认3000
sendMessageTimeout: 5000
# 发送消息失败重试次数，默认2
retryTimesWhenSendFailed: 2
# 异步消息重试此处，默认2
retryTimesWhenSendAsyncFailed: 2
# 消息最大长度，默认1024 * 1024 * 4(默认4M)
maxMessageSize: 4096
# 压缩消息阈值，默认4k(1024 * 4)
compressMessageBodyThreshold: 4096
# 是否在内部发送失败时重试另一个broker，默认false
retryNextServer: false
consumer:
group: consumer-group # 消费者组名

4.3、生产者

@Service
public class RocketMQProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage(String topic, String message) {
        rocketMQTemplate.convertAndSend(topic, message);
        System.out.println("Message sent: " + message);
    }
}

4.4、消费者

@Service
@RocketMQMessageListener(topic = "your-topic", consumerGroup = "consumer-group")
public class RocketMQConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

4.5、接口

    @GetMapping("/send")
    public String sendMessage(@RequestParam String topic, @RequestParam String message) {
        rocketMQProducer.sendMessage(topic, message);
        return "Message sent: " + message;
    }