一、MQ介绍

1.1 为什么要用MQ

消息队列是一种先进先出的数据结构

 MQ

其应用场景主要包含以下3个方面

1.1.1 应用解耦

系统的耦合性越高，容错性就越低。以电商应用为例，用户创建订单后，如果耦合度调用库存系统、物流系统、支付系统，任何一个子系统出了故障或因为升级等原因暂时不可用，都会造成下单操作异常，影响用户使用体验。

电商系统

使用消息队列解耦，系统的容错性就提高了。比如物流系统发生故障，需要几分钟才能来修复，在这段时间内，物流系统要处理的数据被缓存到消息队列中，用户的下单操作正常完成。当物流系统恢复后，补充处理存在消息队列中的订单消息即可，终端系统感知不到物流系统发生过几分钟故障。

电商系统+MQ

1.1.2 流量削峰

常规系统

应用系统如果遇到系统请求流量的瞬间猛增，有可能会将系统压垮。有了消息队列可以将大量请求缓存起来，分散到很长一段时间处理，这样可以大大提高系统的稳定性和用户体验。

MQ流量削峰

一般情况，为了保证系统的稳定性，如果系统负载超过阀值，就会阻止用户请求，这会影响用户体验，而如果使用消息队列将请求缓存起来，等待系统处理完毕后通知用户下单完毕，这样总比不能下单体验要好。

另外，出于经济考虑目的，业务系统正常时段的QPS如果是1000，流量最高峰是10000，为了应对流量高峰配置高性能的服务器显然不划算，这时可以使用消息队列对峰值流量削峰。

1.1.3 数据分发

系统调用

通过消息队列可以让数据在多个系统之间更加流通。数据的生产方不需要关心谁来使用数据，只需要将数据发送到消息队列，数据使用放直接在消息队列中获取数据即可。

MQ数据分发

1.2 MQ的优点和缺点

1.2.1 优点

解耦、削峰、数据分发

1.2.2 缺点

• 系统的可用性降低：系统引入的外部依赖越多，系统稳定性越差。一旦MQ宕机，就会对业务造成影响。如何保证MQ的高可用？

• 系统复杂度提高：MQ的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过MQ进行异步调用。如何保证消息没有被重复消费（幂等性）？怎么处理消息丢失情况？怎么保证消息传递的顺序性？

• 一致性问题：A系统处理完业务，通过MQ给B、C、D三个系统发送数据，如果B系统、C系统处理成功，D系统处理失败。如何保证消息数据处理的一致性？

1.3 各种MQ产品的比较

常见的MQ产品包括：Kafka、ActiveMQ、RabbitMQ、RocketMQ

MQ比较

二、Ubuntu安装RocketMQ

官网安装文档

2.1 RocketMQ环境要求

1) 64bit OS,linux/Unix/Max
2) 64bit JDK 1.8+
3) Maven 3.2.x
4) Git

2.2 Ubuntu安装openjdk8

先检查是否已安装JDK

java -version

如果未安装JDK，需要先安装JDK

# 1、更新软件包
sudo apt-get update

# 2、安装openjdk
sudo apt-get install openjdk-8-jdk

# 3、查看安装版本号
java -version

4、配置JAVA环境

依次运行下述代码找到java的位置

[root@xxx]# which java
/usr/bin/java
[root@xxx]# ls -lrt /usr/bin/java
lrwxrwxrwx. 1 root root 22 7月  23 14:43 /usr/bin/java -> /etc/alternatives/java
[root@xxx]# ls -lrt /etc/alternatives/java
lrwxrwxrwx. 1 root root 73 7月  23 14:43 /etc/alternatives/java -> /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java

找到java路径后，去掉最后的/jre/bin/java，在profile文件中进行修改

vim  /etc/profile

在末尾追加

JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
JRE_HOME=$JAVA_HOME/jre
CLASS_PATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib
PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin
export JAVA_HOME JRE_HOME CLASS_PATH PATH

使配置生效

source /etc/profile

2.3 Ubuntu安装maven

先检查是否已安装maven

mvn -v

如果未安装maven，需要先安装maven

# 1、安装maven
sudo apt-get install maven

# 2、查看版本号
mvn -v

2.4 Ubuntu安装git

先检查是否已安装git

git --version

如果未安装git，需要先安装git

# 1、安装git
sudo apt-get install git

# 2、查看版本号
git --version

2.5 下载RocketMQ并构建

2.5.1 创建data目录，并进入该目录

mkdir /data
cd /data

2.5.2 下载并构建

git clone https://github.com/apache/incubator-rocketmq.git

cd incubator-rocketmq

mvn -Prelease-all -DskipTests clean install -U

2.5.3 配置环境变量

# 修改配置文件
vim /etc/profile

# 在结尾处添加：
export ROCKETMQ_HOME=/data/incubator-rocketmq/distribution/target/rocketmq-4.9.3-SNAPSHOT/rocketmq-4.9.3-SNAPSHOT
export PATH=$JAVA_HOME/bin:$ROCKETMQ_HOME/bin:$PATH


# 使配置生效
source /etc/profile

注意：ROCKETMQ_HOME要配置成你自己的目录

2.5.4 修改JAVA_HOME 路径

修改两个文件：runbroker.sh和runserver.sh
我的路径是/data/incubator-rocketmq/distribution/target/rocketmq-4.9.3-SNAPSHOT/rocketmq-4.9.3-SNAPSHOT/bin/runbroker.sh

修改JAVA_HOME路径

[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64

2.5.5 启动NameServer

RocketMQ NameServer 默认端口9876

# 进入RocketMQ目录
cd /data/incubator-rocketmq/distribution/target/rocketmq-4.9.3-SNAPSHOT/rocketmq-4.9.3-SNAPSHOT

# 启动NameServer
nohup sh bin/mqnamesrv > /data/mqnamesrv.log 2>&1 &

# 查看是否启动成功
netstat -tunlp|grep 9876

# 查看日志
tail -f /data/mqnamesrv.log

root@iZm5eetszs07500os8erolZ:~# cd /data/incubator-rocketmq/distribution/target/rocketmq-4.9.3-SNAPSHOT/rocketmq-4.9.3-SNAPSHOT
root@iZm5eetszs07500os8erolZ:/data/incubator-rocketmq/distribution/target/rocketmq-4.9.3-SNAPSHOT/rocketmq-4.9.3-SNAPSHOT# nohup sh bin/mqnamesrv > /data/mqnamesrv.log 2>&1 &
[1] 2814
root@iZm5eetszs07500os8erolZ:/data/incubator-rocketmq/distribution/target/rocketmq-4.9.3-SNAPSHOT/rocketmq-4.9.3-SNAPSHOT# netstat -tunlp|grep 9876
tcp        0      0 0.0.0.0:9876            0.0.0.0:*               LISTEN      2830/java
root@iZm5eetszs07500os8erolZ:/data/incubator-rocketmq/distribution/target/rocketmq-4.9.3-SNAPSHOT/rocketmq-4.9.3-SNAPSHOT# tail -f /data/mqnamesrv.log
nohup: ignoring input
OpenJDK 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
OpenJDK 64-Bit Server VM warning: UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release.
The Name Server boot success. serializeType=JSON

备注：官网的启动命令是nohup sh bin/mqnamesrv &，但会提示nohup: ignoring input and appending output to 'nohup.out'，所以将命令换成了nohup sh bin/mqnamesrv > /data/mqnamesrv.log 2>&1 &

2.5.6 修改虚拟机内存

RocketMQ默认使用的虚拟机内存较大，启动Broker如果内存不足会失败，需要编辑如下两个配置文件，修改JVM内存大小

vi runbroker.sh

vi runserver.sh

2.5.7 启动Broker

三、RocketMQ简介

RocketMQ是阿里巴巴2016年开源的MQ中间件，使用Java语言开发，在阿里内部，RocketMQ承接了例如双11等高并发场景的消息流转，能够处理万亿级别的消息。

四、RocketMQ基本概念

官网-基本概念

4.1 消息模型（Message Model）

RocketMQ主要由 Producer、Broker、Consumer 三部分组成，其中Producer 负责生产消息，Consumer 负责消费消息，Broker 负责存储消息。Broker 在实际部署过程中对应一台服务器，每个 Broker 可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的 Broker。Message Queue 用于存储消息的物理地址，每个Topic中的消息地址存储于多个 Message Queue 中。ConsumerGroup 由多个Consumer 实例构成。

4.2 消息生产者（Producer）

负责生产消息，一般由业务系统负责生产消息。一个消息生产者会把业务应用系统里产生的消息发送到broker服务器。RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。同步和异步方式均需要Broker返回确认信息，单向发送不需要。

4.3 消息消费者（Consumer）

负责消费消息，一般是后台系统负责异步消费。一个消息消费者会从Broker服务器拉取消息、并将其提供给应用程序。从用户应用的角度而言提供了两种消费形式：拉取式消费、推动式消费。

4.4 主题（Topic）

表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是RocketMQ进行消息订阅的基本单位。

4.5 代理服务器（Broker Server）

消息中转角色，负责存储消息、转发消息。代理服务器在RocketMQ系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。

4.6 名字服务（Name Server）

名称服务充当路由消息的提供者。生产者或消费者能够通过名字服务查找各主题相应的Broker IP列表。多个Namesrv实例组成集群，但相互独立，没有信息交换。

4.7 拉取式消费（Pull Consumer）

Consumer消费的一种类型，应用通常主动调用Consumer的拉消息方法从Broker服务器拉消息、主动权由应用控制。一旦获取了批量消息，应用就会启动消费过程。

4.8 推动式消费（Push Consumer）

Consumer消费的一种类型，该模式下Broker收到数据后会主动推送给消费端，该消费模式一般实时性较高。

4.9 生产者组（Producer Group）

同一类Producer的集合，这类Producer发送同一类消息且发送逻辑一致。如果发送的是事务消息且原始生产者在发送之后崩溃，则Broker服务器会联系同一生产者组的其他生产者实例以提交或回溯消费。

4.10 消费者组（Consumer Group）

同一类Consumer的集合，这类Consumer通常消费同一类消息且消费逻辑一致。消费者组使得在消息消费方面，实现负载均衡和容错的目标变得非常容易。要注意的是，消费者组的消费者实例必须订阅完全相同的Topic。RocketMQ 支持两种消息模式：集群消费（Clustering）和广播消费（Broadcasting）。

4.11 集群消费（Clustering）

集群消费模式下,相同Consumer Group的每个Consumer实例平均分摊消息。

4.12 广播消费（Broadcasting）

广播消费模式下，相同Consumer Group的每个Consumer实例都接收全量的消息。

4.13 普通顺序消息（Normal Ordered Message）

普通顺序消费模式下，消费者通过同一个消息队列（ Topic 分区，称作 Message Queue） 收到的消息是有顺序的，不同消息队列收到的消息则可能是无顺序的。

4.14 严格顺序消息（Strictly Ordered Message）

严格顺序消息模式下，消费者收到的所有消息均是有顺序的。

4.15 消息（Message）

消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。RocketMQ中每个消息拥有唯一的Message ID，且可以携带具有业务标识的Key。系统提供了通过Message ID和Key查询消息的功能。

4.16 标签（Tag）

为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。

五、RocketMQ特性

官网-特性

5.1 订阅与发布

消息的发布是指某个生产者向某个topic发送消息；消息的订阅是指某个消费者关注了某个topic中带有某些tag的消息，进而从该topic消费数据。

5.2 消息顺序

消息有序指的是一类消息消费时，能按照发送的顺序来消费。例如：一个订单产生了三条消息分别是订单创建、订单付款、订单完成。消费时要按照这个顺序消费才能有意义，但是同时订单之间是可以并行消费的。RocketMQ可以严格的保证消息有序。

顺序消息分为全局顺序消息与分区顺序消息，全局顺序是指某个Topic下的所有消息都要保证顺序；部分顺序消息只要保证每一组消息被顺序消费即可。

• 全局顺序 对于指定的一个 Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费。 适用场景：性能要求不高，所有的消息严格按照 FIFO 原则进行消息发布和消费的场景。

• 分区顺序 对于指定的一个 Topic，所有消息根据 sharding key 进行区块分区。 同一个分区内的消息按照严格的 FIFO 顺序进行发布和消费。 Sharding key 是顺序消息中用来区分不同分区的关键字段，和普通消息的 Key 是完全不同的概念。 适用场景：性能要求高，以 sharding key 作为分区字段，在同一个区块中严格的按照 FIFO 原则进行消息发布和消费的场景。

5.3 消息过滤

RocketMQ的消费者可以根据Tag进行消息过滤，也支持自定义属性过滤。消息过滤目前是在Broker端实现的，优点是减少了对于Consumer无用消息的网络传输，缺点是增加了Broker的负担、而且实现相对复杂。

5.4 消息可靠性

RocketMQ支持消息的高可靠，影响消息可靠性的几种情况：

1、Broker非正常关闭
2、Broker异常Crash
3、OS Crash
4、机器掉电，但是能立即恢复供电情况
5、机器无法开机（可能是cpu、主板、内存等关键设备损坏）
磁盘设备损坏。

1)、2)、3)、4) 四种情况都属于硬件资源可立即恢复情况，RocketMQ在这四种情况下能保证消息不丢，或者丢失少量数据（依赖刷盘方式是同步还是异步）。

5)、6)属于单点故障，且无法恢复，一旦发生，在此单点上的消息全部丢失。RocketMQ在这两种情况下，通过异步复制，可保证99%的消息不丢，但是仍然会有极少量的消息可能丢失。通过同步双写技术可以完全避免单点，同步双写势必会影响性能，适合对消息可靠性要求极高的场合，例如与Money相关的应用。注：RocketMQ从3.0版本开始支持同步双写。

5.5 至少一次

至少一次(At least Once)指每个消息必须投递一次。Consumer先Pull消息到本地，消费完成后，才向服务器返回ack，如果没有消费一定不会ack消息，所以RocketMQ可以很好的支持此特性。

5.6 回溯消费

回溯消费是指Consumer已经消费成功的消息，由于业务上需求需要重新消费，要支持此功能，Broker在向Consumer投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度，例如由于Consumer系统故障，恢复后需要重新消费1小时前的数据，那么Broker要提供一种机制，可以按照时间维度来回退消费进度。RocketMQ支持按照时间回溯消费，时间维度精确到毫秒。

5.7 事务消息

RocketMQ事务消息（Transactional Message）是指应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致。

5.8 定时(延时)消息

定时消息（延迟队列）是指消息发送到broker后，不会立即被消费，等待特定时间投递给真正的topic。 broker有配置项messageDelayLevel，默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”，18个level。可以配置自定义messageDelayLevel。注意，messageDelayLevel是broker的属性，不属于某个topic。发消息时，设置delayLevel等级即可：msg.setDelayLevel(level)。level有以下三种情况：

• level == 0，消息为非延迟消息

• 1<=level<=maxLevel，消息延迟特定时间，例如level==1，延迟1s

• level > maxLevel，则level== maxLevel，例如level==20，延迟2h，即最大延时时间为2小时。

定时消息会暂存在名为SCHEDULE_TOPIC_XXXX的topic中，并根据delayTimeLevel存入特定的queue，queueId = delayTimeLevel – 1，即一个queue只存相同延迟的消息，保证具有相同发送延迟的消息能够顺序消费。broker会调度地消费SCHEDULE_TOPIC_XXXX，将消息写入真实的topic。

需要注意的是，定时消息会在第一次写入和调度写入真实topic时都会计数，因此发送数量、tps都会变高。

5.8.2 延时消息的使用场景

比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

扩展：Rabbitmq本身是没有延迟队列的，只能通过Rabbitmq本身的特性来实现，想要Rabbitmq实现延时队列，需要使用Rabbitmq的死信交换机（Exchage）和消息存活时间TTL（Time To Live）

5.8.3 延时消息的使用限制

// org/apache/rocketmq/store/config/MessageStoreConfig.java

private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级（商业版本支持自定义时间），从1s到2h分别对应着等级1到18 消息消费失败会进入延时消息队列，消息发送时间与设置的延时等级和重试次数有关，详见代码SendMessageProcessor.java

5.9 消息重试

Consumer消费消息失败后，要提供一种重试机制，令消息再消费一次。Consumer消费消息失败通常可以认为有以下几种情况：

• 由于消息本身的原因，例如反序列化失败，消息数据本身无法处理（例如话费充值，当前消息的手机号被注销，无法充值）等。这种错误通常需要跳过这条消息，再消费其它消息，而这条失败的消息即使立刻重试消费，99%也不成功，所以最好提供一种定时重试机制，即过10秒后再重试。
• 由于依赖的下游应用服务不可用，例如db连接不可用，外系统网络不可达等。遇到这种错误，即使跳过当前失败的消息，消费其他消息同样也会报错。这种情况建议应用sleep 30s，再消费下一条消息，这样可以减轻Broker重试消息的压力。

RocketMQ会为每个消费组都设置一个Topic名称为“%RETRY%+consumerGroup”的重试队列（这里需要注意的是，这个Topic的重试队列是针对消费组，而不是针对每个Topic设置的），用于暂时保存因为各种异常而导致Consumer端无法消费的消息。考虑到异常恢复起来需要一些时间，会为重试队列设置多个重试级别，每个重试级别都有与之对应的重新投递延时，重试次数越多投递延时就越大。RocketMQ对于重试消息的处理是先保存至Topic名称为“SCHEDULE_TOPIC_XXXX”的延迟队列中，后台定时任务按照对应的时间进行Delay后重新保存至“%RETRY%+consumerGroup”的重试队列中。

RocketMQ消息重试机制：采用时间衰减的方式，使用了自身定时消费的能力。首次在10秒后重试消费，如果消费成功则不再重试，如果消费失败则继续重试消费，第二次载30s后重试消费，以此类推，每次重试的间隔时间都会加长，直到超过最大重试次数（默认16次），则写入死信队列不再重试，重试消费过程中的间隔时间使用了定时消费，重试的消息数据并非直接写入重试队列，而是先写入定时消费队列，再通过定时消息的功能转发到重试队列。（1小时后重试第15次，2小时后重试第16次）

5.10 消息重投

生产者在发送消息时，同步消息失败会重投，异步消息有重试，oneway没有任何保证。消息重投保证消息尽可能发送成功、不丢失，但可能会造成消息重复，消息重复在RocketMQ中是无法避免的问题。消息重复在一般情况下不会发生，当出现消息量大、网络抖动，消息重复就会是大概率事件。另外，生产者主动重发、consumer负载变化也会导致重复消息。如下方法可以设置消息重试策略：

• retryTimesWhenSendFailed:同步发送失败重投次数，默认为2，因此生产者会最多尝试发送retryTimesWhenSendFailed + 1次。不会选择上次失败的broker，尝试向其他broker发送，最大程度保证消息不丢。超过重投次数，抛出异常，由客户端保证消息不丢。当出现RemotingException、MQClientException和部分MQBrokerException时会重投。

• retryTimesWhenSendAsyncFailed:异步发送失败重试次数，异步重试不会选择其他broker，仅在同一个broker上做重试，不保证消息不丢。

• retryAnotherBrokerWhenNotStoreOK:消息刷盘（主或备）超时或slave不可用（返回状态非SEND_OK），是否尝试发送到其他broker，默认false。十分重要消息可以开启。

5.11 流量控制

生产者流控，因为broker处理能力达到瓶颈；消费者流控，因为消费能力达到瓶颈。

生产者流控：

• commitLog文件被锁时间超过osPageCacheBusyTimeOutMills时，参数默认为1000ms，返回流控。

• 如果开启transientStorePoolEnable == true，且broker为异步刷盘的主机，且transientStorePool中资源不足，拒绝当前send请求，返回流控。

• broker每隔10ms检查send请求队列头部请求的等待时间，如果超过waitTimeMillsInSendQueue，默认200ms，拒绝当前send请求，返回流控。

• broker通过拒绝send 请求方式实现流量控制。

注意，生产者流控，不会尝试消息重投。

消费者流控：

• 消费者本地缓存消息数超过pullThresholdForQueue时，默认1000。
• 消费者本地缓存消息大小超过pullThresholdSizeForQueue时，默认100MB。
• 消费者本地缓存消息跨度超过consumeConcurrentlyMaxSpan时，默认2000。

消费者流控的结果是降低拉取频率。

5.12 死信队列

死信队列用于处理无法被正常消费的消息。当一条消息初次消费失败，消息队列会自动进行消息重试；达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消息队列 不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。

RocketMQ将这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），将存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。在RocketMQ中，可以通过使用console控制台对死信队列中的消息进行重发来使得消费者实例再次进行消费。

六、RocketMQ最佳实践

官网-最佳实践

6.1 生产者

6.1.1 Tags的使用

一个应用尽可能用一个Topic，而消息子类型则可以用tags来标识。tags可以由应用自由设置，只有生产者在发送消息设置了tags，消费方在订阅消息时才可以利用tags通过broker做消息过滤

6.1.2 Keys的使用

每个消息在业务层面的唯一标识码要设置到keys字段，方便将来定位消息丢失问题。服务器会为每个消息创建索引（哈希索引），应用可以通过topic、key来查询这条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证key尽可能唯一，这样可以避免潜在的哈希冲突。

6.1.3 消息发送失败处理方式

Producer的send方法本身支持内部重试，重试逻辑如下：

至多重试2次。
如果同步模式发送失败，则轮转到下一个Broker，如果异步模式发送失败，则只会在当前Broker进行重试。这个方法的总耗时时间不超过sendMsgTimeout设置的值，默认10s。
如果本身向broker发送消息产生超时异常，就不会再重试。
以上策略也是在一定程度上保证了消息可以发送成功。如果业务对消息可靠性要求比较高，建议应用增加相应的重试逻辑：比如调用send同步方法发送失败时，则尝试将消息存储到db，然后由后台线程定时重试，确保消息一定到达Broker。

上述db重试方式为什么没有集成到MQ客户端内部做，而是要求应用自己去完成，主要基于以下几点考虑：首先，MQ的客户端设计为无状态模式，方便任意的水平扩展，且对机器资源的消耗仅仅是cpu、内存、网络。其次，如果MQ客户端内部集成一个KV存储模块，那么数据只有同步落盘才能较可靠，而同步落盘本身性能开销较大，所以通常会采用异步落盘，又由于应用关闭过程不受MQ运维人员控制，可能经常会发生 kill -9 这样暴力方式关闭，造成数据没有及时落盘而丢失。第三，Producer所在机器的可靠性较低，一般为虚拟机，不适合存储重要数据。综上，建议重试过程交由应用来控制。

6.1.4 选择oneway形式发送

通常消息的发送是这样一个过程：

• 客户端发送请求到服务器
• 服务器处理请求
• 服务器向客户端返回应答

所以，一次消息发送的耗时时间是上述三个步骤的总和，而某些场景要求耗时非常短，但是对可靠性要求并不高，例如日志收集类应用，此类应用可以采用oneway形式调用，oneway形式只发送请求不等待应答，而发送请求在客户端实现层面仅仅是一个操作系统系统调用的开销，即将数据写入客户端的socket缓冲区，此过程耗时通常在微秒级。

6.2 消费者

6.2.1 消费过程幂等

RocketMQ无法避免消息重复（Exactly-Once），所以如果业务对消费重复非常敏感，务必要在业务层面进行去重处理。可以借助关系数据库进行去重。首先需要确定消息的唯一键，可以是msgId，也可以是消息内容中的唯一标识字段，例如订单Id等。在消费之前判断唯一键是否在关系数据库中存在。如果不存在则插入，并消费，否则跳过。（实际过程要考虑原子性问题，判断是否存在可以尝试插入，如果报主键冲突，则插入失败，直接跳过）

msgId一定是全局唯一标识符，但是实际使用中，可能会存在相同的消息有两个不同msgId的情况（消费者主动重发、因客户端重投机制导致的重复等），这种情况就需要使业务字段进行重复消费。

6.2.2 消费速度慢的处理方式

1、提高消费并行度

绝大部分消息消费行为都属于 IO 密集型，即可能是操作数据库，或者调用 RPC，这类消费行为的消费速度在于后端数据库或者外系统的吞吐量，通过增加消费并行度，可以提高总的消费吞吐量，但是并行度增加到一定程度，反而会下降。所以，应用必须要设置合理的并行度。 如下有几种修改消费并行度的方法：

• 同一个 ConsumerGroup 下，通过增加 Consumer 实例数量来提高并行度（需要注意的是超过订阅队列数的 Consumer 实例无效）。可以通过加机器，或者在已有机器启动多个进程的方式。
• 提高单个 Consumer 的消费并行线程，通过修改参数 consumeThreadMin、consumeThreadMax实现。

2、批量方式消费

某些业务流程如果支持批量方式消费，则可以很大程度上提高消费吞吐量，例如订单扣款类应用，一次处理一个订单耗时 1 s，一次处理 10 个订单可能也只耗时 2 s，这样即可大幅度提高消费的吞吐量，通过设置 consumer的 consumeMessageBatchMaxSize 返个参数，默认是 1，即一次只消费一条消息，例如设置为 N，那么每次消费的消息数小于等于 N。

3、跳过非重要消息

发生消息堆积时，如果消费速度一直追不上发送速度，如果业务对数据要求不高的话，可以选择丢弃不重要的消息。例如，当某个队列的消息数堆积到100000条以上，则尝试丢弃部分或全部消息，这样就可以快速追上发送消息的速度。

4、优化每条消息消费过程

举例如下，某条消息的消费过程如下：

1、根据消息从 DB 查询【数据 1】
2、根据消息从 DB 查询【数据 2】
3、复杂的业务计算
4、向 DB 插入【数据 3】
5、向 DB 插入【数据 4】

这条消息的消费过程中有4次与 DB的 交互，如果按照每次 5ms 计算，那么总共耗时 20ms，假设业务计算耗时 5ms，那么总过耗时 25ms，所以如果能把 4 次 DB 交互优化为 2 次，那么总耗时就可以优化到 15ms，即总体性能提高了 40%。所以应用如果对时延敏感的话，可以把DB部署在SSD硬盘，相比于SCSI磁盘，前者的RT会小很多。

6.3 消费打印日志

如果消息量较少，建议在消费入口方法打印消息，消费耗时等，方便后续排查问题。

七、RocketMQ样例

官网-样例

八、Rocketmq-console控制台

1. 下载开源的rocketmq-externals项目进行部署

https://github.com/apache/rocketmq-externals

2、控制台界面如下

Rocketmq-console