RocketMQ快速入门

2.1 消息生产和消费介绍

使用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,//消息回滚

UNKNOW, //未知状态，一般用于处理超时等现象

}

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

2.2 RocketMQ普通消息生产者

2.2.1 工程创建

我们先实现一个最基本的消息发送，先创建一个springboot工程，工程名字叫rocketmq-demo1

pom.xml

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 " target="_blank">http://maven.apache.org/xsd/maven-4.0.0.xsd">

<modelVersion>4.0.0</modelVersion>

<parent>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-parent</artifactId>

<version>2.0.5.RELEASE</version>

</parent>

<groupId>org.mentu</groupId>

<artifactId>rocketmq-demo1</artifactId>

<version>1.0-SNAPSHOT</version>

<packaging>jar</packaging>

<properties>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

<java.version>1.8</java.version>

<rocketmq.version>4.4.0</rocketmq.version>

</properties>

<dependencies>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-actuator</artifactId>

</dependency>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-test</artifactId>

<scope>test</scope>

</dependency>

<dependency>

<groupId>org.apache.rocketmq</groupId>

<artifactId>rocketmq-client</artifactId>

<version>${rocketmq.version}</version>

</dependency>

</dependencies>

</project>

2.2.2 消息发送

消息发送有这么几个步骤：

创建DefaultMQProducer

设置Namesrv地址

开启DefaultMQProducer

创建消息Message

发送消息

关闭DefaultMQProducer

我们创建一个Producer类，按照上面步骤实现消息发送，代码如下：

在这里插入图片描述

public class Producer {

//指定namesrv地址

private static String NAMESRV_ADDRESS = "192.168.211.143:9876";

public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {

//创建一个DefaultMQProducer,需要指定消息发送组

DefaultMQProducer producer = new DefaultMQProducer("Test_Quick_Producer_Name");

//指定Namesvr地址

producer.setNamesrvAddr(NAMESRV_ADDRESS);

//启动Producer

producer.start();

//创建消息

Message message = new Message(

"Test_Quick_Topic", //主题

"TagA", //标签，可以用来做过滤

"KeyA", //唯一标识，可以用来查找消息

"hello rocketmq".getBytes() //要发送的消息字节数组

);

//发送消息

SendResult result = producer.send(message);

//关闭producer

producer.shutdown();

}

}

我们可以在控制台查看到对应的消息,控制台地址：http://localhost:8080/#/message 我们可以看到如下消息：在这里插入图片描述

注意：这里时间查询以消息存储时间为准，注意服务器的时间有可能不准确。

2.3 RocketMQ普通消息消费者

2.3.1 消息消费

消费者消费消息有这么几个步骤：

创建DefaultMQPushConsumer

设置namesrv地址

设置subscribe，这里是要读取的主题信息

创建消息监听MessageListener

获取消息信息

返回消息读取状态

创建Consumer类，按照上面步骤实现消息消费，代码如下：

在这里插入图片描述

上图代码如下：

public class Consumer {

//指定namesrv地址

private static String NAMESRV_ADDRESS = "192.168.211.143:9876";

public static void main(String[] args) throws MQClientException {

//创建DefaultMQPushConsumer

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Test_Quick_Consumer_Name");

//设置namesrv地址

consumer.setNamesrvAddr(NAMESRV_ADDRESS);

//设置要读取的topic

consumer.subscribe(

"Test_Quick_Topic", //指定要读取的消息主题

"TagA"); //指定要读取的消息过滤信息,多个标签数据，则可以输入"tag1 || tag2 || tag3"

//创建消息监听

consumer.setMessageListener(new MessageListenerConcurrently() {

@Override

public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

try {

//获取第1个消息

MessageExt message = msgs.get(0);

//获取主题

String topic = message.getTopic();

//获取标签

String tags = message.getTags();

//获取消息

String result = new String(message.getBody(),"UTF-8");

System.out.println("topic:"+topic+",tags:"+tags+",result:"+result);

} catch (UnsupportedEncodingException e) {

e.printStackTrace();

//消息重试

return ConsumeConcurrentlyStatus.RECONSUME_LATER;

}

//消息消费成功

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

}

});

//启动消费监听

consumer.start();

}

}

控制台运行结果

topic:Test_Quick_Topic,tags:TagA,result:hello rocketmq

2.4 RocketMQ顺序消息

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

如何保证顺序

在MQ的模型中，顺序需要由3个阶段去保障：

消息被发送时保持顺序

消息被存储时保持和发送的顺序一致

消息被消费时保持和存储的顺序一致

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

在这里插入图片描述

2.4.1 消息生产者

我们创建一个消息生产者OrderProducer,这里每次发消息都会发到同一个队列中，代码如下：

在这里插入图片描述

上图代码如下：

public class OrderProducer {

//nameserver地址

private static String namesrvaddress="192.168.211.143:9876;";

public static void main(String[] args) throws UnsupportedEncodingException, InterruptedException, RemotingException, MQClientException, MQBrokerException {

//创建DefaultMQProducer

DefaultMQProducer producer = new DefaultMQProducer("order_producer_group_name");

//设置namesrv地址

producer.setNamesrvAddr(namesrvaddress);

//启动Producer

producer.start();

//创建消息

Message message = new Message(

"Topic_Order_Demo",

"TagOrder",

"KeyOrder",

"hello order message!".getBytes(RemotingHelper.DEFAULT_CHARSET));

//发送消息

SendResult result = producer.send(

message, //要发送的消息

new MessageQueueSelector() {

@Override

public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {

return mqs.get((Integer) arg);

}

},

1);//设置存入第几个队列中,这里是下标，从0开始

//关闭Producer

producer.shutdown();

}

}

2.4.2 消息消费者

创建一个消息消费者OrderConsumer,消息监听用MessageListenerOrderly来实现顺序消息，代码如下：

在这里插入图片描述

上图代码如下：

public class OrderConsumer {

//nameserver地址

private static String namesrvaddress="192.168.211.143:9876;";

public static void main(String[] args) throws MQClientException {

//创建消息消费对象DefaultMQConsumer

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_consumer_group_name");

//设置nameserver地址

consumer.setNamesrvAddr(namesrvaddress);

//设置消费顺序

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

//设置消息拉取最大数

consumer.setConsumeMessageBatchMaxSize(5);

//设置消费主题

consumer.subscribe("Topic_Order_Demo","TagOrder");

//消息监听

consumer.setMessageListener(new MessageListenerOrderly() {

@Override

public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {

try {

for (MessageExt msg : msgs) {

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 ConsumeOrderlyStatus.SUCCESS;

}

});

//启动Consumer

consumer.start();

}

}

我们打开控制台，可以看到消息发送到了第2个队列中了。

在这里插入图片描述

提示：大家在测试顺序消息的时候，可以将上面消息生产者连续发送10个或者更多来测试顺序。

2.5 RocketMQ事务消息

在RocketMQ4.3.0版本后，开放了事务消息这一特性，对于分布式事务而言，最常说的还是二阶段提交协议。

2.5.1 RocketMQ事务消息流程

RocketMQ的事务消息，主要是通过消息的异步处理，可以保证本地事务和消息发送同时成功执行或失败，从而保证数据的最终一致性，这里我们先看看一条事务消息从诞生到结束的整个时间线流程：

在这里插入图片描述

事务消息的成功投递是需要经历三个Topic的，分别是：

Half Topic：用于记录所有的prepare消息

Op Half Topic：记录已经提交了状态的prepare消息

Real Topic：事务消息真正的Topic,在Commit后会才会将消息写入该Topic，从而进行消息的投递

2.5.2 事务消息生产者

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

监听器TransactionListenerImpl，代码如下：在这里插入图片描述

上图代码如下：

public class TransactionListenerImpl implements TransactionListener {

//存储当前线程对应的事务状态

private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

/***

* 发送prepare消息成功后回调该方法用于执行本地事务

* @param msg:回传的消息，利用transactionId即可获取到该消息的唯一Id

* @param arg:调用send方法时传递的参数，当send时候若有额外的参数可以传递到send方法中，这里能获取到

* @return

*/

@Override

public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {

//获取线程ID

String transactionId = msg.getTransactionId();

//初始状态为0

localTrans.put(transactionId,0);

try {

//此处执行本地事务操作

System.out.println("....执行本地事务");

Thread.sleep(70000);

System.out.println("....执行完成本地事务");

} catch (InterruptedException e) {

e.printStackTrace();

//发生异常，则回滚消息

localTrans.put(transactionId,2);

return LocalTransactionState.UNKNOW;

}

//修改状态

localTrans.put(transactionId,1);

System.out.println("executeLocalTransaction------状态为1");

//本地事务操作如果成功了，则提交该消息，让该消息可见

return LocalTransactionState.UNKNOW;

}

/***

* 消息回查

* @param msg

* @return

*/

@Override

public LocalTransactionState checkLocalTransaction(MessageExt msg) {

//获取事务id

String transactionId = msg.getTransactionId();

//通过事务id获取对应的本地事务执行状态

Integer status = localTrans.get(transactionId);

System.out.println("消息回查-----"+status);

switch (status){

case 0:

return LocalTransactionState.UNKNOW;

case 1:

return LocalTransactionState.COMMIT_MESSAGE;

case 2:

return LocalTransactionState.ROLLBACK_MESSAGE;

}

return LocalTransactionState.UNKNOW;

}

}

创建消息发送对象TransactionProducer,代码如下：

在这里插入图片描述

上图代码如下：

public class TransactionProducer {

//nameserver地址

private static String namesrvaddress="192.168.211.143:9876;";

public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, InterruptedException {

//创建事务消息发送对象

TransactionMQProducer producer = new TransactionMQProducer("transaction_producer_group_name");

//设置namesrv地址

producer.setNamesrvAddr(namesrvaddress);

//创建监听器

TransactionListener transactionListener = new TransactionListenerImpl();

//创建线程池

ExecutorService executorService = new ThreadPoolExecutor(

2,

5,

100,

TimeUnit.SECONDS,

new ArrayBlockingQueue<Runnable>(

2000),

new ThreadFactory() {

@Override

public Thread newThread(Runnable runnable) {

Thread thread = new Thread(runnable);

thread.setName("client-transaction-msg-check-thread");

return thread;

}

}

);

//设置线程池

producer.setExecutorService(executorService);

//设置监听器

producer.setTransactionListener(transactionListener);

//启动producer

producer.start();

//创建消息

Message message = new Message(

"TopicTxt_Demo",

"TagTx",

"KeyTx1",

"hello".getBytes(RemotingHelper.DEFAULT_CHARSET));

//发送事务消息,此时消息不可见

TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(message, "发送消息，回传所需数据！");

System.out.println(transactionSendResult);

//休眠

Thread.sleep(120000);

//关闭

producer.shutdown();

}

}

2.5.3 事务消息

事务消息的消费者和普通消费者一样，这里我们就不做介绍了，直接贴代码：

public class TransactionConsumer {

//nameserver地址

private static String namesrvaddress="192.168.211.143:9876;";

public static void main(String[] args) throws MQClientException {

//创建DefaultMQPushConsumer

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("transaction_consumer_group_name");

//设置nameserver地址

consumer.setNamesrvAddr(namesrvaddress);

//设置每次拉去的消息个数

consumer.setConsumeMessageBatchMaxSize(5);

//设置消费顺序

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

//设置监听的消息

consumer.subscribe("TopicTxt_Demo","TagTx");

//消息监听

consumer.registerMessageListener(new MessageListenerConcurrently() {

@Override

public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

try {

for (MessageExt msg : msgs) {

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.CONSUME_SUCCESS;

}

});

//启动消费

consumer.start();

}

}

事务消息参考地址：http://rocketmq.apache.org/docs/transaction-example/

2.5.4 RocketMQ实现分布式事务流程

MQ事务消息解决分布式事务问题，但第三方MQ支持事务消息的中间件不多，比如RocketMQ，他们支持事务消息的方式也是类似于采用的二阶段提交，但是市面上一些主流的MQ都是不支持事务消息的，比如 RabbitMQ 和 Kafka 都不支持。

以阿里的 RocketMQ 中间件为例，其思路大致为：

第一阶段Prepared消息，会拿到消息的地址。

第二阶段执行本地事务，第三阶段通过第一阶段拿到的地址去访问消息，并修改状态。

也就是说在业务方法内要想消息队列提交两次请求，一次发送消息和一次确认消息。如果确认消息发送失败了RocketMQ会定期扫描消息集群中的事务消息，这时候发现了Prepared消息，它会向消息发送者确认，所以生产方需要实现一个check接口，RocketMQ会根据发送端设置的策略来决定是回滚还是继续发送确认消息。这样就保证了消息发送与本地事务同时成功或同时失败。

在这里插入图片描述

2.6 消息广播/批量发送

上面发送消息，我们测试的时候，可以发现消息只有一个消费者能收到，如果我们想实现消息广播，让每个消费者都能收到消息也是可以实现的。而且上面发送消息的时候，每次都是发送单条Message对象，能否批量发送呢？答案是可以的。

2.6.1 消息生产者

创建消息生产者BroadcastingProducer,代码如下：

在这里插入图片描述

上图代码如下：

public class BroadcastingProducer {

//nameserver地址

private static String namesrvaddress="192.168.211.143:9876;";

public static void main(String[] args) throws UnsupportedEncodingException, MQClientException, RemotingException, InterruptedException, MQBrokerException {

//创建DefaultMQProducer

DefaultMQProducer producer = new DefaultMQProducer("broadcasting_producer_group");

//指定nameserver地址

producer.setNamesrvAddr(namesrvaddress);

//启动

producer.start();

//创建消息

List<Message> messages = new ArrayList<Message>();

for (int i = 0; i <20 ; i++) {

Message message = new Message(

"Topic_broadcasting",

"TagBroad",

"KeyBroad"+i,

(i+"--hello brodcasting").getBytes(RemotingHelper.DEFAULT_CHARSET));

//将消息添加到集合中

messages.add(message);

}

//批量发送消息

producer.send(messages);

//关闭

producer.shutdown();

}

}

2.6.2 消息消费者

广播消费模式其实就是每个消费者都能读取到消息，我们这里只需要将消费者的消费模式设置成广播模式即可。consumer.setMessageModel(MessageModel.BROADCASTING);,代码如下：

在这里插入图片描述

上图代码如下：

public class BroadcastingConsumerDemo1 {

//广播模式

private static String namesrvaddress="192.168.211.143:9876;";

public static void main(String[] args) throws MQClientException {

//创建DefaultMQConsumer

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("broadcasting_consumer_group");

//指定nameserver地址

consumer.setNamesrvAddr(namesrvaddress);

//指定要消费的消息主体

consumer.subscribe("Topic_broadcasting","*");

//指定消费顺序

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

//指定一次拉取条数

consumer.setConsumeMessageBatchMaxSize(2);

//指定消费模式 集群模式/广播模式

consumer.setMessageModel(MessageModel.BROADCASTING);

//创建监听，监听消息

consumer.setMessageListener(new MessageListenerConcurrently() {

@Override

public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

for (MessageExt msg : msgs) {

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("demo1 topic:"+topic+",tags:"+tags+",keys:"+keys+",body:"+body);

} catch (UnsupportedEncodingException e) {

e.printStackTrace();

}

}

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

}

});

//启动

consumer.start();

}

}

运行测试，可以发现每个节点都消费了20条消息。效果如下图：

在这里插入图片描述

学习：

https://blog.csdn.net/Same_Liu/article/details/89517571?spm=1001.2014.3001.5506