RocketMQ 是阿里巴巴开源的一款 分布式消息中间件，具有高吞吐、低延迟、高可用等特点，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。
一.RockerMQ的概念：
RocketMQ主要由 Producer、Broker、Consumer 三部分组成，其中Producer 负责生产消息，Consumer 负责消费消息，Broker 负责存储消息。Broker 在实际部署过程中对应一台服务器，每个 Broker 可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的 Broker。Message Queue 用于存储消息的物理地址，每个Topic中的消息地址存储于多个 Message Queue 中。ConsumerGroup 由多个Consumer 实例构成。Rockermq官网链接地址
1. 架构分析：
- 生产者（Producer）：负责产生消息，生产者向消息服务器发送由业务应用程序系统生成的消息。
- 消费者（Consumer）：负责消费消息，消费者从消息服务器拉取信息并将其输入用户应用程序。
- 消息服务器（Broker）：是消息存储中心，主要作用是接收来自 Producer 的消息并存储， Consumer 从这里取得消息。
- 名称服务器（NameServer）：用来保存 Broker 相关 Topic 等元信息并给 Producer ，提供 Consumer 查找 Broker 信息。
2.整体流程：
1、首先启动 Namesrv（名称服务器），Namesrv 起来后监听端口，等待 Broker、Producer、Consumer 连上来，相当于一个路由控制中心。
2、Broker 启动，跟所有的 Namesrv 保持长连接，定时发送心跳包。
心跳包中，包含当前 Broker 信息(IP+端口等)以及存储所有 Topic 信息。 注册成功后，Namesrv 集群中就有 Topic 跟 Broker 的映射关系。
3、收发消息前，先创建 Topic 。创建 Topic 时，需要指定该 Topic 要存储在哪些 Broker上。也可以在发送消息时自动创建Topic。
4、Producer 发送消息。
启动时，先跟 Namesrv 集群中的其中一台建立长连接，并从Namesrv 中获取当前发送的 Topic 存在哪些 Broker 上，然后跟对应的 Broker 建立长连接，直接向 Broker 发消息。
5、Consumer 消费消息。
Consumer 跟 Producer 类似。跟其中一台 Namesrv 建立长连接，获取当前订阅 Topic 存在哪些 Broker 上，然后直接跟 Broker 建立连接通道，开始消费消息。
二.Rockermq单机部署：
1.安装解压包：
rocketmq-all-4.6.0-source-release.zip
点击横杠即可下载安装包。
2.Windows部署 / Linux部署：
（1）Windows部署：
解压后使用 Maven 编译 RocketMQ 源码。命令行操作如下：
# 进入 RocketMQ 源码目录
$ cd rocketmq-all-4.6.0-source-release# Maven 编译 RocketMQ ，并跳过测试。耐心等待...
$ mvn -Prelease-all -DskipTests clean install -U
随后进入.\bin\mqnamesrv.cmd来启动Namesrv，如果报错 Java not found，请检查 JAVA_HOME 环境变量。
.\mqnamesrv.cmd正常会出现下图：
随后新建一个CMD窗口（不要关闭 NameServer），进入 bin 目录，启动Broker：
mqbroker.cmd -c ..\conf\broker.conf -n localhost:9876
（2）Linux部署：
使用wget命令下载，随后使用unzip命令解压：
# 下载
$ wget wget http://mirror.bit.edu.cn/apache/rocketmq/4.6.0/rocketmq-all-4.6.0-source-release.zip# 解压
$ unzip rocketmq-all-4.6.0-source-release.zip随后同样使用 Maven 编译 RocketMQ 源码。命令行操作如下：
# 进入 RocketMQ 源码目录
$ cd rocketmq-all-4.6.0-source-release# Maven 编译 RocketMQ ，并跳过测试。耐心等待...
$ mvn -Prelease-all -DskipTests clean install -U
编译完成，在我们进入 distribution 目录下，就可以看到 RocketMQ 的发布包了。命令行操作如下：
# 进入 distribution 目录下
$ cd distribution/target/rocketmq-4.6.0/rocketmq-4.6.0# 打印目录
$ ls
40 -rwxr-xr-x 1 yunai staff 17336 Nov 19 20:59 LICENSE8 -rwxr-xr-x 1 yunai staff 1338 Nov 19 20:59 NOTICE
16 -rwxr-xr-x 1 yunai staff 4225 Nov 19 20:59 README.md0 drwxr-xr-x 6 yunai staff 192 Dec 3 12:48 benchmark # 性能基准测试0 drwxr-xr-x 30 yunai staff 960 Nov 19 20:59 bin # 执行脚本0 drwxr-xr-x 12 yunai staff 384 Nov 19 20:59 conf # 配置文件0 drwxr-xr-x 36 yunai staff 1152 Dec 3 12:48 lib # RocketMQ jar 包随后启动Namesrv：
nohup sh bin/mqnamesrv &
启动完成后，查看日志：
# 查看 Namesrv 日志。
$ tail -f ~/logs/rocketmqlogs/namesrv.log2019-12-03 12:58:04 INFO main - The Name Server boot success. serializeType=JSON
默认情况下，Namesrv 日志文件所在地址为
~/logs/rocketmqlogs/namesrv.log。如果想要自定义，可以通过conf/logback_namesrv.xml配置文件来进行修改。
随后启动broker：
在
conf目录下，RocketMQ 提供了多种 Broker 的配置文件：
broker.conf：单主，异步刷盘。2m/：双主，异步刷盘。2m-2s-async/：两主两从，异步复制，异步刷盘。2m-2s-sync/：两主两从，同步复制，异步刷盘。dledger/：Dledger 集群，至少三节点。
这里，我们只启动一个 RocketMQ Broker 服务，所以使用 broker.conf 配置文件。命令行操作如下：
nohup sh bin/mqbroker -c conf/broker.conf -n 127.0.0.1:9876 &
通过
-c参数，配置读取的主 Broker 配置。通过
-n参数，设置 RocketMQ Namesrv 地址。
3.测试消息：
（1）测试发送消息：
通过使用 bin/tools.sh 工具类，实现测试发送消息。命令行操作如下：
# 设置 Namesrv 服务器的地址
export NAMESRV_ADDR=127.0.0.1:9876# 执行生产者 Producer 发送测试消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
如果发送成功，我们会看到大量成功的发送日志。
SendResult [sendStatus=SEND_OK, msgId=FE800000000000004F2B5386138462F500000D7163610D67E7F100F4, offsetMsgId=C0A8032C00002A9F000000000000D7EE, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=0], queueOffset=61]
SendResult [sendStatus=SEND_OK, msgId=FE800000000000004F2B5386138462F500000D7163610D67E7F200F5, offsetMsgId=C0A8032C00002A9F000000000000D8D1, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=1], queueOffset=61]
通过发送结果为 sendStatus=SEND_OK 状态，说明消息都发送成功了。
（2）测试消费消息：
通过使用 bin/tools.sh 工具类，实现测试消费消息。命令行操作如下：
# 设置 Namesrv 服务器的地址
export NAMESRV_ADDR=127.0.0.1:9876# 执行消费者 Consumer 消费测试消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
如果消费成功，我们会看到大量成功的消费日志。
ConsumeMessageThread_4 Receive New Messages: [MessageExt [queueId=2, storeSize=227, queueOffset=131, sysFlag=0, bornTimestamp=1575354513732, bornHost=/192.168.3.44:55510, storeTimestamp=1575354513733, storeHost=/192.168.3.44:10911, msgId=C0A8032C00002A9F000000000001D1FC, commitLogOffset=119292, bodyCRC=1549304357, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=145, CONSUME_START_TIME=1575354867104, UNIQ_KEY=FE800000000000004F2B5386138462F500000D7163610D67E944020E, CLUSTER=DefaultCluster, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 53, 50, 54], transactionId='null'}]]
ConsumeMessageThread_3 Receive New Messages: [MessageExt [queueId=2, storeSize=227, queueOffset=130, sysFlag=0, bornTimestamp=1575354513729, bornHost=/192.168.3.44:55510, storeTimestamp=1575354513729, storeHost=/192.168.3.44:10911, msgId=C0A8032C00002A9F000000000001CE70, commitLogOffset=118384, bodyCRC=1530218044, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=145, CONSUME_START_TIME=1575354867103, UNIQ_KEY=FE800000000000004F2B5386138462F500000D7163610D67E941020A, CLUSTER=DefaultCluster, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 53, 50, 50], transactionId='null'}]]
通过 ConsumeMessageThread_4 和 ConsumeMessageThread_3 线程名，我们可以看出，目前是进行并发消费消息。
三.Rockermq的简单使用实例：
（1）项目结构：
rocketmq-demo/
├── src/
│ ├── main/
│ │ ├── java/
│ │ │ └── com/example/
│ │ │ ├── producer/OrderProducer.java # 生产者
│ │ │ ├── consumer/OrderConsumer.java # 消费者
│ │ │ └── RocketmqDemoApplication.java # 启动类
│ │ └── resources/
│ │ └── application.yml # 配置文件
│ └── test/
└── pom.xml # Maven依赖（2）添加依赖：
<dependencies><!-- Spring Boot Starter --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><!-- RocketMQ Starter --><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.3</version></dependency>
</dependencies>（3）编写配置文件：
rocketmq:name-server: localhost:9876 # NameServer地址producer:group: order-producer-group # 生产者组consumer:group: order-consumer-group # 消费者组topic: order-topic # 订阅的Topic（4）编写生产者代码：
package com.example.producer;import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;@Component
public class OrderProducer {@Autowiredprivate RocketMQTemplate rocketMQTemplate;public void sendOrderMessage(String orderId) {// 发送消息（Topic: order-topic, Tag: pay）rocketMQTemplate.convertAndSend("order-topic:pay", "订单支付成功，订单ID: " + orderId);System.out.println("生产者发送消息: " + orderId);}
}（5）编写消费者代码：
package com.example.consumer;import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;@Component
@RocketMQMessageListener(topic = "order-topic", // 监听的TopicselectorExpression = "pay", // 过滤Tag=pay的消息consumerGroup = "order-consumer-group"
)
public class OrderConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println("消费者收到消息: " + message);}
}（6）启动类：
package com.example;import com.example.producer.OrderProducer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;@SpringBootApplication
public class RocketmqDemoApplication {public static void main(String[] args) {ConfigurableApplicationContext context = SpringApplication.run(RocketmqDemoApplication.class, args);// 测试发送消息OrderProducer producer = context.getBean(OrderProducer.class);producer.sendOrderMessage("1001"); // 发送订单ID=1001}
}
