中国建设银行网站企业网银广州网站快速优化排名

前言

工作中很多种场景用到消息队列，消息队列简单来说就是消息的传输过程中保存消息的一种容器。项目中引入消息队列中间件主要解决了异步处理、应用耦合、流量削峰等问题。今天我们来学习一下阿里开源的一款产品 RocketMQ。

一、RocketMQ简介

RocketMQ 是一款低延迟、高并发、高可用、高可靠的分布式消息中间件。具备异步通信的优势，系统拓扑简单、上下游耦合较弱，主要应用于异步解耦，流量削峰填谷等场景。

1.1 整体架构

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

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

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

● NameServer： NameServer 是整个 RocketMQ 的大脑，相当于路由/注册和发现中心。可使用集群方式部署，集群中各个NameServer 都是无状态的即无法感知其它 NameServer 的存在。NameServer 的主要作用是为消息生产者和消息消费者提供有关 Topic 的路由信息，所以 NameServer 就需要存储路由信息，并且能够管理 Broker 节点，包括路由注册、路由删除等功能。

● Broker： RocketMQ 核心组件之一，大部分重量级工作都是通过 Broker 来完成的。Borker 处理各种请求和存储消息，决定整个 RocketMQ 体系的吞吐性能、可靠性和可用性。Broker 可以存储多个 Topic 的消息，每个 Topic 的消息也可以分片存储于不同的 Broker。MessageQueue 用于存储消息的物理地址，每个 Topic 中的消息地址存储于多个 MessageQueue 中。

主题（Topic）
表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是 RocketMQ 进行消息订阅的基本单位。
在RocketMQ中，每个Topic默认都会有4个队列，并且每个队列都有一个id，默认从0开始，依次递增

二、RocketMQ安装部署

2.1 RocketMQ 下载

安装部署RocketMQ前, 需要先安装JDK，这里不做记录!

RocketMQ 下载链接：https://archive.apache.org/dist/rocketmq/4.9.1/

下载到本地后 win + r 打开 cmd 输入如下命令将文件上传到 Liunx 服务器的 /mydata/rocketmq/目录。

scp rocketmq-all-4.9.1-bin-release.zip的位置 root@192.168.57.129:/mydata/rocketmq/

输入 Linux 开机密码后上传成功，如下图：

或者直接使用如下命令下载安装包

wget https://archive.apache.org/dist/rocketmq/4.9.1/rocketmq-all-4.9.1-bin-release.zip

解压安装包

unzip rocketmq-all-4.9.1-bin-release.zip

重命名

mv rocketmq-all-4.9.1-bin-release rocketmq-4.9.1

进入rocketmq-4.9.1 文件夹

cd rocketmq-4.9.1

查看 rocketmq-4.9.1 目录

2.2 修改 JVM 参数

在启动 RocketMQ 之前，建议修改启动时的 JVM 参数，因为默认的参数都比较大，为了避免内存不够，建议修改小。

修改 runserver.sh 的 JVM 参数

vi /mydata/rocketmq/rocketmq-4.9.1/bin/runserver.sh

将原来的初始堆内存和最大最内存调整为 512m，esc + : wq 保存后退出

修改 runbroker.sh 的 JVM 参数

vi /mydata/rocketmq/rocketmq-4.9.1/bin/runbroker.sh

2.3 启动 NameServer 和 Broker

启动 nameServer

nohup /mydata/rocketmq/rocketmq-4.9.1/bin/mqnamesrv -n 192.168.57.129:9876 &

查看 nohup.out 日志

[root@localhost bin]# cat nohup.out 
Java HotSpot(TM) 64-Bit Server VM warning: Using the DefNew young collector with the CMS collector is deprecated and will likely be removed in a future release
Java HotSpot(TM) 64-Bit Server VM warning: UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release.
The Name Server boot success. serializeType=JSON

编写 broker.conf 配置文件

vim /mydata/rocketmq/rocketmq-4.9.1/conf/broker.conf

加上这行，开启自动创建 Topic

autoCreateTopicEnable = true

启动 broker

nohup /mydata/rocketmq/rocketmq-4.9.1/bin/mqbroker -n 192.168.57.129:9876 &

查看 nohup.out 日志

[root@localhost bin]# cat nohup.out 
Java HotSpot(TM) 64-Bit Server VM warning: Using the DefNew young collector with the CMS collector is deprecated and will likely be removed in a future release
Java HotSpot(TM) 64-Bit Server VM warning: UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release.
The Name Server boot success. serializeType=JSON
The broker[localhost.localdomain, 192.168.57.129:10911] boot success. serializeType=JSON and name server is 192.168.57.129:9876

2.4 验证发送和接受消息

● 配置 nameserver 的环境变量，在发送和接受消息之前，需要告诉客户端 nameserver 的位置。配置 NAMESRV_ADDR 如下。

vim /etc/profile

配置信息

export NAMESRV_ADDR=192.168.57.129:9876

重新加载配置，让它生效

source /etc/profile

● 使用 bin/tools.sh 工具验证消息的发送和接收，默认会发 1000 条数据。

启动生产者发送消息

tools.sh org.apache.rocketmq.example.quickstart.Producer

启动消费者接收消息

tools.sh org.apache.rocketmq.example.quickstart.Consumer

2.5 停止 NameServer 和 Broker

停止 nameserver

[root@localhost bin]# mqshutdown namesrv
The mqnamesrv(11783) is running...
Send shutdown request to mqnamesrv(11783) OK

停止 broker

[root@localhost bin]# mqshutdown broker
The mqbroker(12178) is running...
Send shutdown request to mqbroker(12178) OK

2.6 配置全局环境

vim /etc/profile

export ROCKETMQ_HOME=/mydata/rocketmq/rocketmq-4.9.1
export PATH=$PATH:$ROCKETMQ_HOME/bin

source /etc/profile

这样就不必每次进入 RocketMQ 的安装目录了，直接可以使用 mqnamesrv 和mqbroker 指令。

三、RocketMQ应用场景

3.1 异步处理

如下图，简单模拟用户下单流程，若使用同步通信方式用户下订单后，后台需要执行创建订单，扣减库存，加积分，生成优惠卷四个业务模块。通常来说它们都是以微服务的形式部署在不同机器上，网络不可靠因素无疑增加了用户下单失败的风险，调用链长增加了用户下单后返回下单成功的时长。

使用消息队列后，当用户点击下订单操作，后台往消息队列发送消息(userId，productId，…)，后续创建订单，扣减库存，加积分，优惠卷四个业务只需要去消息队列的某个 Topic 订阅消息，当 Broker 的某个 Topic 有它们所订阅的消息后就能接受消息，进而去执行它们的业务逻辑。所以用户下订单后便能很快返下单成功。

相比第一种同步调用方案，使用 MQ 异步处理能明显提高了系统的吞吐量，而且上下游的依赖关系明显减弱了，达到了解耦的目的。

3.2 应用解耦

如下图所示：当用户发送请求经过 API 网关路由到服务 A，服务 A 再发送消息给 MQ 的某个 Topic ，服务 B，服务 C，服务 D 订阅这个 Topic 后就能接受服务 A 的消息，再去处理各自的业务，达到应用解耦的目的。

绝大多数电商业务场景都使用消息队列来解决类似的系统耦合过于紧密的问题。引入消息队列后订单服务在订单变化时发送一条消息到消息队列的一个主题 Order 中，所有下游系统都订阅主题 Order，这样每个下游系统都可以获得一份实时完整的订单数据无论增加、减少下游系统或是下游系统需求如何变化，订单服务都无需做任何更改，实现了订单服务与下游服务的解耦。

3.3 流量削峰

如何避免过多的请求压垮我们的秒杀系统?

设计思路: 使用消息队列隔离网关和后端服务，以达到流量控制和保护后端服务的目的。

加入消息队列之后，整个秒杀的流程如下：

1. 网关在收到请求后，将请求放入到 MQ 中
2. 后端服务从请求 MQ 获取请求，完成后续秒杀处理过程，返回响应
   

代价:

1. 增加系统调用链的环节，导致总体的响应时延变长
2. 同步调用变成了异步调用，增加系统的复杂度
3. 成本问题，MQ高可能、高可用

常见限流算法:

1. 固定窗口算法
2. 滑动窗口算法
3. 漏桶算法
4. 令牌桶算法

令牌桶控制流量的原理是: 单位时间内只发放固定数量的令牌到令牌桶中，规定服务在处理请求之前必须先从令牌桶中拿出一个令牌，如果令牌桶中没有令牌，则拒绝请求。这样就保证单位时间内，能处理的请求不超过发放令牌的数量，起到了流量控制的作用。

更多的使用场景:

1. 通过MQ实现分布式事务，最终一致性
2. 作为发布/订阅系统实现一个微服务级系统间的观察者模式
3. 连接流计算任务和数据
4. 用于将消息广播给大量接收者，数据同步

备注： 应用解耦和流量控制部分来自B站作者(架构驿站)的消息队列进阶课。在此表示感谢！