一. 简介

MQ全称Message Queue，消息队列(MQ)是一种应用程序对应用程序的通信方式，应用程序通过读写出入队列的消息(针对应用程序的数据)来通信，而无需专用连接来链接它们，消息传递指的是程序之间通过在消息中发送数据进行通信。对立诶的使用除去了接收和发送应用程序同时执行的要求。

RocketMQ的前身是Metaq，当Metag3.0发布时，产品名称改为RocketMQ，RocketMQ是一款分布式、队列模型的消息中间件，具有以下特点：

• 能够保证严格的消息顺序
• 提供丰富的消息拉取模式
• 高效的订阅者水平扩展能力
• 实时的消息订阅机制
• 支持事务消息
• 亿级消息堆积能力

二. 单节点RocketMQ安装

2.1 环境准备

我们先准备一台centos虚拟机，ip：**，在hosts文件中配置地址与IP的映射关系。

IP	hostname	mastername
**	Rocketmq-nameserver1	Rocketmq-master1

修改/etc/hosts文件，加入如下映射关系

** rocketmq-nameserver1
** rocketmq-master1

2.2 安装配置

我们可以把安装文件上传到虚拟机上，并解压安装

解压文件存放到/usr/local/server/mq目录下

unzip rocketmq-all-4.6.1-bin-release.zip -d /usr/local/server/mq

更新解压后的文件名

mv rocketmq-all-4.6.1-bin-release rocketmq

创建RocketMQ存储文件的目录，执行如下命令

[root@localhost rocketmq]# mkdir logs
[root@localhost rocketmq]# mkdir store
[root@localhost rocketmq]# cd store
[root@localhost store]# mkdir commitlog
[root@localhost store]# mkdir consumerqueue
[root@localhost store]# mkdir index

文件夹说明

logs：存储RocketMQ日志记录
store：存储RocketMQ数据文件目录
commitlog：存储RocketMQ消息信息
consumerqueue、index：存储消息的索引数据

conf目录配置文件说明

2m-2s-async：2主2从异步
2m-2s-sync ：2主2从同步
2m-noslave ：2主没有从

我们这里先配置简单节点，可以修改2m-2s-async的配置实现。

进入2m-2s-async目录，修改第一个配置文件broker-a.properties

vi broker-a.properties

将如下配置覆盖掉broker-a.properties所有配置

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示Master，>0表示slave
brokerId=0
#nameServer地址，分号分隔
namesrvAddr=rocketmq-nameserver:9876
#在发送消息时，自动创建服务器不存在的Topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许Broker自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许Broker自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker对外服务的监听端口
listenPort=10911
#删除文件时间点，默认是凌晨4点
deleteWhen=04
#文件保留时间，默认48小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumerQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumerQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/server/mq/rocketmq/store
#commitLog存储路径
storePathCommitLog=/usr/local/server/mq/rocketmq/store/commitlog
#消息队列存储路径
storePathConsumerQueue=/usr/local/server/mq/rocketmq/store/consumerqueue
#消息索引存储路径
storePathIndex=/usr/local/server/mq/rocketmq/store/index
#checkpoint文件存储路径
storeCheckpoint=/usr/local/server/mq/rocketmq/store/checkpoint
#abort文件存储路径
abortFile=/usr/local/server/mq/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumerQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumerQueueThorougnInterval=60000
#Broker的角色
# - ASYNC_MASTER 异步复制Master
# - SYNC_MASTER  同步双写Master
# - SLAVE
brokerRole=ASYNC_MASTER
#刷盘方式
# - ASYNC_FLUSH 异步刷盘
# - SYNC_FLUSH  同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128

进入conf目录，替换所有xml中的${user.name}，保证日志路径正确

sed -i 's#${user.home}#/usr/local/server/mq/rocketmq#g' *.xml

注意：sed -i 在这里起一个批量替换的作用

sed -i 's#原字符串#新字符#g' 替换的文件

RocketMQ对内存要求比较高，最少1G，如果内存太少，会影响RocketMQ的运行效率和执行性能，我们需要修改bin目录小的runbroker.sh和runserver.sh文件

runbroker.sh

改前：
JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g Xmn4g"
改后：
JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g Xmn1g"

runserver.sh

改前：
JAVA_OPT="${JAVA_OPT} -server Xms4g Xmx4g Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
改后：
JAVA_OPT="${JAVA_OPT} -server Xms1g Xmx1g Xmn1g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

先启动namesrv

nohup sh mqnamesrv &

再启动broker

nohup sh mqbroker -c /usr/local/server/mq/rocketmq/conf/2m-2s-async/broker-a.properties > /dev/null 2>&1 &

输入jps查看进程

[root@localhost rocketmqlogs]# jps
3225 NamesrvStartup
3290 BrokerStartup
3454 Jps

三. 概念模型

• Name Server 是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
• Broker部署相对复杂，Broker分为 Master与Slave，一个 Master可以对应多个Slave，但是一个Slave 只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave，Master也可以部署多个，每个Broker与Name Server 集群中的所有节点建立长连接，定时注册 Topic信息到所有Name Server。
• Producer与Name Server集群中的其中一个节点（随机选择）建立长连接，定期从Name Server取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。
• Consumer与Name Server集群中的其中一个节点（随机选择）建立长连接，定期从Name Server取Topic路由信息，并向提供Topic服务的 Master、Slave 建立长连接，且定时向Master、Slave 发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。

三. 生产者与消费者

使用RocketMQ可以发送普通消息、顺序消息、事务消息，顺序消息能实现有序消费，事务消息可以解决分布式事务实现数据最终一致性。

RocketMQ有2中常见的消费模式，分别是DefaultMQPushConsumer和DefaultMQPullConsumer模式，这2种模式字面理解一个是推送消息，一个是拉取消息，这里有个误区，其实无论是Push还是Pull，其本质都是拉取消息，只是实现机制不一样。

DefaultMQPushConsumer其实并不是broker主动向consumer推送消息，而是consumer向broker发出请求，保持了一种长连接，broker会每5秒检测一次是否有消息，如果有消息，则将消息推送给consumer。使用DefaultMQPushConsumer实现消费，broker会主动记录消息消费的偏移量。

DefaultMQPullConsumer是消费方主动去broker拉取数据，一般会在本地使用定时任务实现，使用它获得消息状态方便、负载均衡性能可控，但消息的及时性差，而且需要手动记录消息消费的偏移量信息，所以在实际应用场景中推荐使用Push方式。

RocketMQ发送的消息默认会存储到4个队列中去，当然创建几个队列存储数据，可以自己定义。

RocketMQ作为MQ消息中间件，ack机制必不可少，在RocketMQ中常见的应答状态如下：

LocalTransactionState:主要针对事务消息的应答状态public enum LocalTransactionState {// 消息提交COMMIT_MESSAGE,// 消息回滚ROLLBACK_MESSAGE,// 未知状态，一般用于处理超时等现象UNKNOWN;
}

ConsumeConcurrentlyStatus:主要针对消息消费的应答状态public enum ConsumeConcurrentlyStatus {// 消息消费成功CONSUME_SUCCESS,// 消息重试，一般消息消费失败后，RocketMQ为了保障数据的可靠性，具有重试机制RECONSUME_LATER;
}

重发时间是：(broker.log中有)

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

3.1 生产者代码实现

流程如下：

1. 创建DefaultMQProducer
2. 设置Namesrv地址
3. 开启DefaultMQProducer
4. 创建消息Message
5. 发送消息
6. 关闭DefaultMQProducer

public class Producer {public static void main(String[] args) throws Exception {// 1、创建DefaultMQProducerDefaultMQProducer producer = new DefaultMQProducer("demo_producer_group");// 2、设置Namesrv地址producer.setNamesrvAddr("");// 3、开启DefaultMQProducerproducer.start();// 4、创建消息MessageMessage message = new Message("Topic_Demo", "Tags", "Keys_1", "Hello RocketMQ".getBytes());// 5、发送消息SendResult result = producer.send(message);System.out.println(result);// 6、关闭DefaultMQProducerproducer.shutdown();}
}

3.2 消费者代码实现

流程如下：

1. 创建DefaultMQPushConsumer
2. 设置namesrv地址
3. 设置subscribe，这里是要读取订单主题信息
4. 创建消息监听MessageListener
5. 返回消息读取状态
6. 启动consumer

public class Consumer {public static void main(String[] args) throws MQClientException {// 1、创建DefaultMQPushConsumerDefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo_consumer_group");// 2、设置namesrv地址consumer.setNamesrvAddr("IP:9876");// 3、设置消息拉取最大数consumer.setConsumeMessageBatchMaxSize(3);// 4、设置subscribe，这里是要读取订单主题信息// topic：指定消费的主题，subExpression：过滤规则consumer.subscribe("stock", "Tags");// 5、创建消息监听MessageListenerconsumer.setMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {// 迭代消息for(MessageExt msg : msgs){try {//获取主题String topic = msg.getTopic();String tag = msg.getTags();byte[] body = msg.getBody();String result = new String(body, RemotingHelper.DEFAULT_CHARSET);System.out.println("消费的topic:" + topic + "标签是：" + tag + "内容为"+result);} catch (UnsupportedEncodingException e) {e.printStackTrace();return ConsumeConcurrentlyStatus.RECONSUME_LATER;}}// 7、返回消息读取状态// 消息消费成功return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});//启动consumerconsumer.start();}}

3.3 RocketMQ顺序消息发送

消息有序指的是可以按照消息的发送顺序来消费。RocketMQ可以严格的保证消息顺序，但这个顺序，不是全局顺序，只是分区(queue)顺序，要全局顺序只能一个分区。

如何保证顺序

在MQ模型中，顺序需要由3个阶段去保证：
1、消息被发送时保证顺序
2、消息被存储时保持和发送的顺序一致
3、消息被消费时保持和存储的顺序一致

发送时保持顺序意味着对于有顺序要求的消息，用户应该在同一线程中采用同步的方式发送。存储保持和顺序一致则要求在同一线程中被发送出来的消息A和B，存储时在空间上A一定在B之前。而消费保持和存储一致则要求消息A、B到达Consumer之后必须按照先A后B的顺序被处理。

Producer端：

public class Peoducer {public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {// 1、创建DefaultMQProducerDefaultMQProducer producer = new DefaultMQProducer("demo_producer_test");// 2、设置Namesrv地址producer.setNamesrvAddr("");// 3、开启DefaultMQProducerproducer.start();// 5、发送消息// 第一个参数：发送的消息信息// 第二个参数：选择指定的消息队列对象(会将所有的消息队列传进来)// 第三个参数：指定对应的队列下表// 连续发送5条消息for (int i = 0; i < 5; i++) {// 4、创建消息MessageMessage message = new Message("stock", "Tags", "Keys_1" + i, ("Hello RocketMQ"+i).getBytes());SendResult result = producer.send(message, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> list, Message message, Object o) {Integer index = (Integer) o;return list.get(index);}}, 1);System.out.println(result);}// 6、关闭DefaultMQProducerproducer.shutdown();}
}

SendResult [sendStatus=SEND_OK, msgId=C0A80168405418B4AAC26B825AB00000, offsetMsgId=2F5D35C400002A9F000000000006177A, messageQueue=MessageQueue [topic=stock, brokerName=broker-a, queueId=1], queueOffset=12]
SendResult [sendStatus=SEND_OK, msgId=C0A80168405418B4AAC26B825AD30001, offsetMsgId=2F5D35C400002A9F0000000000061834, messageQueue=MessageQueue [topic=stock, brokerName=broker-a, queueId=1], queueOffset=13]
SendResult [sendStatus=SEND_OK, msgId=C0A80168405418B4AAC26B825AE00002, offsetMsgId=2F5D35C400002A9F00000000000618EE, messageQueue=MessageQueue [topic=stock, brokerName=broker-a, queueId=1], queueOffset=14]
SendResult [sendStatus=SEND_OK, msgId=C0A80168405418B4AAC26B825AEE0003, offsetMsgId=2F5D35C400002A9F00000000000619A8, messageQueue=MessageQueue [topic=stock, brokerName=broker-a, queueId=1], queueOffset=15]
SendResult [sendStatus=SEND_OK, msgId=C0A80168405418B4AAC26B825AFC0004, offsetMsgId=2F5D35C400002A9F0000000000061A62, messageQueue=MessageQueue [topic=stock, brokerName=broker-a, queueId=1], queueOffset=16]

Consumer端

public class Consumer {public static void main(String[] args) throws Exception {// 1、创建DefaultMQPushConsumerDefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo_consumer_order_group");// 2、设置namesrv地址consumer.setNamesrvAddr("");// 3、设置消息拉取最大数consumer.setConsumeMessageBatchMaxSize(2);// 4、设置subscribe，这里是要读取订单主题信息// topic：指定消费的主题，subExpression：过滤规则consumer.subscribe("Topic_Order_Demo","*");// 5、创建消息监听MessageListenerconsumer.setMessageListener(new MessageListenerOrderly() {@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext consumeOrderlyContext) {// 迭代消息for(MessageExt msg : msgs){try {//获取主题String topic = msg.getTopic();String tag = msg.getTags();byte[] body = msg.getBody();String result = new String(body, RemotingHelper.DEFAULT_CHARSET);System.out.println("Order Consumer消费信息----topic: " + topic + ", tags: " + tag + ", result: " + result);                    } catch (UnsupportedEncodingException e) {e.printStackTrace();return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;}}return ConsumeOrderlyStatus.SUCCESS;}});// 开启Consumerconsumer.start();}}

Order Consumer消费信息----topic: stock, tags: Tags, result: Hello RocketMQ0
Order Consumer消费信息----topic: stock, tags: Tags, result: Hello RocketMQ1
Order Consumer消费信息----topic: stock, tags: Tags, result: Hello RocketMQ2
Order Consumer消费信息----topic: stock, tags: Tags, result: Hello RocketMQ3
Order Consumer消费信息----topic: stock, tags: Tags, result: Hello RocketMQ4

3.4 事务性消息

3.4.1 概述

在RocketMQ4.3.0版本后，开放了事务消息这一特性，对于分布式事务而言，最常说的还是两阶段提交协议。RocketMQ的事务消息，主要是通过消息的异步处理，可以保证本地事务和消息发送同时成功执行或失败，从而保证数据的最终一致性，这里我们先看看一条事务消息从诞生到结束的整个时间线流程：

3.4.2 事务性生产者代码实现

我们创建一个事务消息生产者TransactionProducer，事务消息发送消息对象是TransactionMQProducer，为了实现本地事务操作和回查，我们需要创建一个监听器，监听器需要实现TransactionListener接口，实现步骤如下：

1. 创建TransactionMQProducer
2. 设置Namesrv地址
3. 指定消息监听对象，用于执行本地事务和消息回查
4. 指定线程池
5. 开启TransactionMQProducer
6. 创建消息Message
7. 发送事务消息
8. 关闭TransactionMQProducer

TransactionProducer代码如下：

public class TracsactionProducer {public static void main(String[] args) throws Exception {// 1、创建TransactionMQProducerTransactionMQProducer producer = new TransactionMQProducer("demo_producer_transaction_group");// 2、设置Namesrv地址producer.setNamesrvAddr("**");// 3、指定消息监听对象，用于执行本地事务和消息回查producer.setTransactionListener(new TransactionListenerImpl());// 4、指定线程池ExecutorService executorService = new ThreadPoolExecutor(2,5,100,TimeUnit.SECONDS,new ArrayBlockingQueue<>(2000),r -> {Thread thread = new Thread(r);thread.setName("client-transaction-msg-check-thread");return thread;});producer.setExecutorService(executorService);// 5、开启TransactionMQProducerproducer.start();// 6、创建消息Message// topic：主题，tags: 标签，主要用于消息过滤，keys：消息的唯一值，body：消息体Message message = new Message("stock","Tags","Keys_T","Hello Transaction RocketMQ Message".getBytes(RemotingHelper.DEFAULT_CHARSET));// 7、发送事务消息// 第一个参数：发送的消息信息// 第二个参数：选择指定的消息队列对象(会将所有的消息队列传进来)TransactionSendResult result = producer.sendMessageInTransaction(message, "hello-transaction");System.out.println(result);// 8、关闭TransactionMQProducerproducer.shutdown();}
}

监听器代码如下：

public class TransactionListenerImpl implements TransactionListener {/*** 存储对应事务的状态信息，key：事务ID，value：当前事务执行的状态*/private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();//执行本地事务@Overridepublic LocalTransactionState executeLocalTransaction(Message message, Object o) {//事务IDString transactionId = message.getTransactionId();// 0：执行中，状态未知，1：本地事务执行成功，2：本地事务执行失败localTrans.put(transactionId, 0);// 业务执行，处理本地事务System.out.println("Hello-----Transaction");try {System.out.println("正在执行本地事务");Thread.sleep(2000);System.out.println("本地事务执行成功");localTrans.put(transactionId, 1);} catch (InterruptedException e) {e.printStackTrace();localTrans.put(transactionId, 2);return LocalTransactionState.ROLLBACK_MESSAGE;}return LocalTransactionState.COMMIT_MESSAGE;}//回查事务状态@Overridepublic LocalTransactionState checkLocalTransaction(MessageExt messageExt) {// 获取事务IDString transactionId = messageExt.getTransactionId();//获取消息状态Integer status = localTrans.get(transactionId);switch (status){case 0:return LocalTransactionState.UNKNOW;case 1:return LocalTransactionState.COMMIT_MESSAGE;case 2:return LocalTransactionState.ROLLBACK_MESSAGE;}return LocalTransactionState.UNKNOW;}
}

3.4.3 事务性消费者代码实现

事务消息的消费者和普通消费者一样，代码如下：

public class TransactionConsumer {public static void main(String[] args) throws Exception {// 1、创建DefaultMQPushConsumerDefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo_consumer_transaction_group");// 2、设置namesrv地址consumer.setNamesrvAddr("");// 3、设置消息拉取最大数consumer.setConsumeMessageBatchMaxSize(2);// 设置消息顺序consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);// 4、设置subscribe，这里是要读取订单主题信息// topic：指定消费的主题，subExpression：过滤规则consumer.subscribe("Topic_Transaction_Demo","*");// 5、创建消息监听MessageListenerconsumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {for (MessageExt msg: list) {try {String topic = msg.getTopic();String tags = msg.getTags();String keys = msg.getKeys();String body = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);System.out.println("topic: " + topic + ", tags: " + tags + ", keys: " + keys + ", body: " + body);} catch (UnsupportedEncodingException e) {e.printStackTrace();return ConsumeConcurrentlyStatus.RECONSUME_LATER;}}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});// 开启Consumerconsumer.start();}
}