RocketMQ的顺序消费和事务消费

╰+哭是因爲堅強的太久メ 2022-05-09 04:22 176阅读 0赞

一、三种消费 ：1.普通消费 2. 顺序消费 3.事务消费

1.1  顺序消费：在网购的时候，我们需要下单，那么下单需要假如有三个顺序，第一、创建订单 ，第二：订单付款，第三：订单完成。也就是这个三个环节要有顺序，这个订单才有意义。RocketMQ可以保证顺序消费，他的实现是生产者（一个生产者可以对多个主题去发送消息）将这个三个消息放在topic（一个topic默认有4个队列）的一个队列里面，单机支持上万个持久化队列，消费端去消费的时候也是只能有一个Consumer去取得这个队列里面的数据，然后顺序消费。

单个节点（Producer端1个、Consumer端1个）

Producer端

Consumer端代码

1.  `package order;`
2.   
3.  `import java.util.List;`
4.  `import java.util.concurrent.TimeUnit;`
5.  `import java.util.concurrent.atomic.AtomicLong;`
6.   
7.  `import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;`
8.  `import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext;`
9.  `import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;`
10. `import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;`
11. `import com.alibaba.rocketmq.client.exception.MQClientException;`
12. `import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;`
13. `import com.alibaba.rocketmq.common.message.MessageExt;`
14.  
15. `/**`
16. `* 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）`
17. `*/`
18. `public class Consumer1 {`
19.  
20. `public static void main(String[] args) throws MQClientException {`
21. `DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_Consumer");`
22. `consumer.setNamesrvAddr("192.168.100.145:9876;192.168.100.146:9876;192.168.100.149:9876;192.168.100.239:9876");`
23.  
24. `/**`
25. `* 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>`
26. `* 如果非第一次启动，那么按照上次消费的位置继续消费`
27. `*/`
28. `consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);`
29.  
30. `consumer.subscribe("TopicOrderTest", "*");`
31.  
32. `consumer.registerMessageListener(new MessageListenerOrderly() {`
33. `AtomicLong consumeTimes = new AtomicLong(0);`
34.  
35. `public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {`
36. `// 设置自动提交`
37. `context.setAutoCommit(true);`
38. `for (MessageExt msg : msgs) {`
39. `System.out.println(msg + ",内容：" + new String(msg.getBody()));`
40. `}`
41.  
42. `try {`
43. `TimeUnit.SECONDS.sleep(5L);`
44. `} catch (InterruptedException e) {`
45.  
46. `e.printStackTrace();`
47. `}`
48. `;`
49.  
50. `return ConsumeOrderlyStatus.SUCCESS;`
51. `}`
52. `});`
53.  
54. `consumer.start();`
55.  
56. `System.out.println("Consumer1 Started.");`
57. `}`
58.  
59. `}`

结果如下图所示：

![Center][]

这个五条数据被顺序消费了

多个节点（Producer端1个、Consumer端2个）

Producer端代码：

1.  `package order;`
2.   
3.  `import java.util.List;`
4.   
5.  `import com.alibaba.rocketmq.client.exception.MQBrokerException;`
6.  `import com.alibaba.rocketmq.client.exception.MQClientException;`
7.  `import com.alibaba.rocketmq.client.producer.DefaultMQProducer;`
8.  `import com.alibaba.rocketmq.client.producer.MessageQueueSelector;`
9.  `import com.alibaba.rocketmq.client.producer.SendResult;`
10. `import com.alibaba.rocketmq.common.message.Message;`
11. `import com.alibaba.rocketmq.common.message.MessageQueue;`
12. `import com.alibaba.rocketmq.remoting.exception.RemotingException;`
13.  
14. `/**`
15. `* Producer，发送顺序消息`
16. `*/`
17. `public class Producer {`
18. `public static void main(String[] args) {`
19. `try {`
20. `DefaultMQProducer producer = new DefaultMQProducer("order_Producer");`
21. `producer.setNamesrvAddr("192.168.100.145:9876;192.168.100.146:9876;192.168.100.149:9876;192.168.100.239:9876");`
22.  
23. `producer.start();`
24.  
25. `// String[] tags = new String[] { "TagA", "TagB", "TagC", "TagD",`
26. `// "TagE" };`
27.  
28. `for (int i = 1; i <= 5; i++) {`
29.  
30. `Message msg = new Message("TopicOrderTest", "order_1", "KEY" + i, ("order_1 " + i).getBytes());`
31.  
32. `SendResult sendResult = producer.send(msg, new MessageQueueSelector() {`
33. `public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {`
34. `Integer id = (Integer) arg;`
35. `int index = id % mqs.size();`
36. `return mqs.get(index);`
37. `}`
38. `}, 0);`
39.  
40. `System.out.println(sendResult);`
41. `}`
42. `for (int i = 1; i <= 5; i++) {`
43.  
44. `Message msg = new Message("TopicOrderTest", "order_2", "KEY" + i, ("order_2 " + i).getBytes());`
45.  
46. `SendResult sendResult = producer.send(msg, new MessageQueueSelector() {`
47. `public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {`
48. `Integer id = (Integer) arg;`
49. `int index = id % mqs.size();`
50. `return mqs.get(index);`
51. `}`
52. `}, 1);`
53.  
54. `System.out.println(sendResult);`
55. `}`
56. `for (int i = 1; i <= 5; i++) {`
57.  
58. `Message msg = new Message("TopicOrderTest", "order_3", "KEY" + i, ("order_3 " + i).getBytes());`
59.  
60. `SendResult sendResult = producer.send(msg, new MessageQueueSelector() {`
61. `public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {`
62. `Integer id = (Integer) arg;`
63. `int index = id % mqs.size();`
64. `return mqs.get(index);`
65. `}`
66. `}, 2);`
67.  
68. `System.out.println(sendResult);`
69. `}`
70.  
71. `producer.shutdown();`
72. `} catch (MQClientException e) {`
73. `e.printStackTrace();`
74. `} catch (RemotingException e) {`
75. `e.printStackTrace();`
76. `} catch (MQBrokerException e) {`
77. `e.printStackTrace();`
78. `} catch (InterruptedException e) {`
79. `e.printStackTrace();`
80. `}`
81. `}`
82. `}`

Consumer1

1.  `/**`
2.  `* 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）`
3.  `*/`
4.  `public class Consumer1 {`
5.   
6.  `public static void main(String[] args) throws MQClientException {`
7.  `DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_Consumer");`
8.  `consumer.setNamesrvAddr("192.168.100.145:9876;192.168.100.146:9876;192.168.100.149:9876;192.168.100.239:9876");`
9.   
10. `/**`
11. `* 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>`
12. `* 如果非第一次启动，那么按照上次消费的位置继续消费`
13. `*/`
14. `consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);`
15.  
16. `consumer.subscribe("TopicOrderTest", "*");`
17.  
18. `/**`
19. `* 实现了MessageListenerOrderly表示一个队列只会被一个线程取到`
20. `*，第二个线程无法访问这个队列`
21. `*/`
22. `consumer.registerMessageListener(new MessageListenerOrderly() {`
23. `AtomicLong consumeTimes = new AtomicLong(0);`
24.  
25. `public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {`
26. `// 设置自动提交`
27. `context.setAutoCommit(true);`
28. `for (MessageExt msg : msgs) {`
29. `System.out.println(msg + ",内容：" + new String(msg.getBody()));`
30. `}`
31.  
32. `try {`
33. `TimeUnit.SECONDS.sleep(5L);`
34. `} catch (InterruptedException e) {`
35.  
36. `e.printStackTrace();`
37. `}`
38. `;`
39.  
40. `return ConsumeOrderlyStatus.SUCCESS;`
41. `}`
42. `});`
43.  
44. `consumer.start();`
45.  
46. `System.out.println("Consumer1 Started.");`
47. `}`
48.  
49. `}`

Consumer2

1.  `/**`
2.  `* 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）`
3.  `*/`
4.  `public class Consumer2 {`
5.   
6.  `public static void main(String[] args) throws MQClientException {`
7.  `DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_Consumer");`
8.  `consumer.setNamesrvAddr("192.168.100.145:9876;192.168.100.146:9876;192.168.100.149:9876;192.168.100.239:9876");`
9.   
10. `/**`
11. `* 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>`
12. `* 如果非第一次启动，那么按照上次消费的位置继续消费`
13. `*/`
14. `consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);`
15.  
16. `consumer.subscribe("TopicOrderTest", "*");`
17.  
18. `/**`
19. `* 实现了MessageListenerOrderly表示一个队列只会被一个线程取到`
20. `*，第二个线程无法访问这个队列`
21. `*/`
22. `consumer.registerMessageListener(new MessageListenerOrderly() {`
23. `AtomicLong consumeTimes = new AtomicLong(0);`
24.  
25. `public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {`
26. `// 设置自动提交`
27. `context.setAutoCommit(true);`
28. `for (MessageExt msg : msgs) {`
29. `System.out.println(msg + ",内容：" + new String(msg.getBody()));`
30. `}`
31.  
32. `try {`
33. `TimeUnit.SECONDS.sleep(5L);`
34. `} catch (InterruptedException e) {`
35.  
36. `e.printStackTrace();`
37. `}`
38. `;`
39.  
40. `return ConsumeOrderlyStatus.SUCCESS;`
41. `}`
42. `});`
43.  
44. `consumer.start();`
45.  
46. `System.out.println("Consumer2 Started.");`
47. `}`
48.  
49. `}`

先启动Consumer1和Consumer2，然后启动Producer，Producer会发送15条消息  
Consumer1消费情况如图，都按照顺序执行了

![Center 1][]  
  
  
Consumer2消费情况如图，都按照顺序执行了

![Center 2][]

二、事务消费

这里说的主要是分布式事物。下面的例子的数据库分别安装在不同的节点上。

事物消费需要先说说什么是事物。比如说：我们跨行转账，从工商银行转到建设银行，也就是我从工商银行扣除1000元之后，我的建设银行也必须加1000元。这样才能保证数据的一致性。假如工商银行转1000元之后，建设银行的服务器突然宕机，那么我扣除了1000，但是并没有在建设银行给我加1000，就出现了数据的不一致。因此加1000和减1000才行，减1000和减1000必须一起成功，一起失败。

再比如，我们进行网购的时候，我们下单之后，订单提交成功，仓库商品的数量必须减一。但是订单可能是一个数据库，仓库数量可能又是在另个数据库里面。有可能订单提交成功之后，仓库数量服务器突然宕机。这样也出现了数据不一致的问题。

使用消息队列来解决分布式事物：

现在我们去外面饭店吃饭，很多时候都不会直接给了钱之后直接在付款的窗口递饭菜，而是付款之后他会给你一张小票，你拿着这个小票去出饭的窗口取饭。这里和我们的系统类似，提高了吞吐量。即使你到第二个窗口，师傅告诉你已经没饭了，你可以拿着这个凭证去退款，即使中途由于出了意外你无法到达窗口进行取饭，但是只要凭证还在，可以将钱退给你。这样就保证了数据的一致性。

如何保证凭证（消息）有2种方法：

1、在工商银行扣款的时候，余额表扣除1000，同时记录日志，而且这2个表是在同一个数据库实例中，可以使用本地事物解决。然后我们通知建设银行需要加1000给该用户，建设银行收到之后给我返回已经加了1000给用户的确认信息之后，我再标记日志表里面的日志为已经完成。

2、通过消息中间件

原文地址：http://www.jianshu.com/p/453c6e7ff81c

![Center 3][]

RocketMQ第一阶段发送Prepared消息时，会拿到消息的地址，第二阶段执行本地事物，第三阶段通过第一阶段拿到的地址去访问消息，并修改消息的状态。

细心的你可能又发现问题了，如果确认消息发送失败了怎么办？RocketMQ会定期扫描消息集群中的事物消息，如果发现了Prepared消息，它会向消息发送端(生产者)确认，Bob的钱到底是减了还是没减呢？如果减了是回滚还是继续发送确认消息呢？RocketMQ会根据发送端设置的策略来决定是回滚还是继续发送确认消息。这样就保证了消息发送与本地事务同时成功或同时失败。

例子：

Consumer 端

1.  `package transaction;`
2.   
3.  `import java.util.List;`
4.   
5.  `import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;`
6.  `import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;`
7.  `import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;`
8.  `import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;`
9.  `import com.alibaba.rocketmq.client.exception.MQClientException;`
10. `import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;`
11. `import com.alibaba.rocketmq.common.message.MessageExt;`
12.  
13. `/**`
14. `* Consumer，订阅消息`
15. `*/`
16. `public class Consumer {`
17.  
18. `public static void main(String[] args) throws InterruptedException, MQClientException {`
19. `DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("transaction_Consumer");`
20. `consumer.setNamesrvAddr("192.168.100.145:9876;192.168.100.146:9876;192.168.100.149:9876;192.168.100.239:9876");`
21. `consumer.setConsumeMessageBatchMaxSize(10);`
22. `/**`
23. `* 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>`
24. `* 如果非第一次启动，那么按照上次消费的位置继续消费`
25. `*/`
26. `consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);`
27.  
28. `consumer.subscribe("TopicTransactionTest", "*");`
29.  
30. `consumer.registerMessageListener(new MessageListenerConcurrently() {`
31.  
32. `public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {`
33.  
34. `try {`
35.  
36. `for (MessageExt msg : msgs) {`
37. `System.out.println(msg + ",内容：" + new String(msg.getBody()));`
38. `}`
39.  
40. `} catch (Exception e) {`
41. `e.printStackTrace();`
42.  
43. `return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试`
44.  
45. `}`
46.  
47. `return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功`
48. `}`
49. `});`
50.  
51. `consumer.start();`
52.  
53. `System.out.println("transaction_Consumer Started.");`
54. `}`
55. `}`

Producer端

1.  `package transaction;`
2.   
3.  `import com.alibaba.rocketmq.client.exception.MQClientException;`
4.  `import com.alibaba.rocketmq.client.producer.SendResult;`
5.  `import com.alibaba.rocketmq.client.producer.TransactionCheckListener;`
6.  `import com.alibaba.rocketmq.client.producer.TransactionMQProducer;`
7.  `import com.alibaba.rocketmq.common.message.Message;`
8.   
9.  `/**`
10. `* 发送事务消息例子`
11. `*`
12. `*/`
13. `public class Producer {`
14. `public static void main(String[] args) throws MQClientException, InterruptedException {`
15.  
16. `TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();`
17. `TransactionMQProducer producer = new TransactionMQProducer("transaction_Producer");`
18. `producer.setNamesrvAddr("192.168.100.145:9876;192.168.100.146:9876;192.168.100.149:9876;192.168.100.239:9876");`
19. `// 事务回查最小并发数`
20. `producer.setCheckThreadPoolMinSize(2);`
21. `// 事务回查最大并发数`
22. `producer.setCheckThreadPoolMaxSize(2);`
23. `// 队列数`
24. `producer.setCheckRequestHoldMax(2000);`
25. `producer.setTransactionCheckListener(transactionCheckListener);`
26. `producer.start();`
27.  
28. `// String[] tags = new String[] { "TagA", "TagB", "TagC", "TagD", "TagE"`
29. `// };`
30. `TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();`
31. `for (int i = 1; i <= 2; i++) {`
32. `try {`
33. `Message msg = new Message("TopicTransactionTest", "transaction" + i, "KEY" + i,`
34. `("Hello RocketMQ " + i).getBytes());`
35. `SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);`
36. `System.out.println(sendResult);`
37.  
38. `Thread.sleep(10);`
39. `} catch (MQClientException e) {`
40. `e.printStackTrace();`
41. `}`
42. `}`
43.  
44. `for (int i = 0; i < 100000; i++) {`
45. `Thread.sleep(1000);`
46. `}`
47.  
48. `producer.shutdown();`
49.  
50. `}`
51. `}`

TransactionExecuterImpl  --执行本地事务

1.  `package transaction;`
2.   
3.  `import com.alibaba.rocketmq.client.producer.LocalTransactionExecuter;`
4.  `import com.alibaba.rocketmq.client.producer.LocalTransactionState;`
5.  `import com.alibaba.rocketmq.common.message.Message;`
6.   
7.  `/**`
8.  `* 执行本地事务`
9.  `*/`
10. `public class TransactionExecuterImpl implements LocalTransactionExecuter {`
11. `// private AtomicInteger transactionIndex = new AtomicInteger(1);`
12.  
13. `public LocalTransactionState executeLocalTransactionBranch(final Message msg, final Object arg) {`
14.  
15. `System.out.println("执行本地事务msg = " + new String(msg.getBody()));`
16.  
17. `System.out.println("执行本地事务arg = " + arg);`
18.  
19. `String tags = msg.getTags();`
20.  
21. `if (tags.equals("transaction2")) {`
22. `System.out.println("======我的操作============，失败了 -进行ROLLBACK");`
23. `return LocalTransactionState.ROLLBACK_MESSAGE;`
24. `}`
25. `return LocalTransactionState.COMMIT_MESSAGE;`
26. `// return LocalTransactionState.UNKNOW;`
27. `}`
28. `}`

TransactionCheckListenerImpl--未决事务，服务器回查客户端(目前已经被阉割啦)

1.  `package transaction;`
2.   
3.  `import com.alibaba.rocketmq.client.producer.LocalTransactionState;`
4.  `import com.alibaba.rocketmq.client.producer.TransactionCheckListener;`
5.  `import com.alibaba.rocketmq.common.message.MessageExt;`
6.   
7.  `/**`
8.  `* 未决事务，服务器回查客户端`
9.  `*/`
10. `public class TransactionCheckListenerImpl implements TransactionCheckListener {`
11. `// private AtomicInteger transactionIndex = new AtomicInteger(0);`
12.  
13. `//在这里，我们可以根据由MQ回传的key去数据库查询，这条数据到底是成功了还是失败了。`
14. `public LocalTransactionState checkLocalTransactionState(MessageExt msg) {`
15. `System.out.println("未决事务，服务器回查客户端msg =" + new String(msg.getBody().toString()));`
16. `// return LocalTransactionState.ROLLBACK_MESSAGE;`
17.  
18. `return LocalTransactionState.COMMIT_MESSAGE;`
19.  
20. `// return LocalTransactionState.UNKNOW;`
21. `}`
22. `}`

producer端：发送数据到MQ，并且处理本地事物。这里模拟了一个成功一个失败。Consumer只会接收到本地事物成功的数据。第二个数据失败了，不会被消费。

![Center 4][]

Consumer只会接收到一个，第二个数据不会被接收到

![Center 5][]

\--------------------- 本文来自 Mr\_蜗牛 的CSDN 博客 ，全文地址请点击：https://blog.csdn.net/u010634288/article/details/57158374?utm\_source=copy

[Center]: /images/20220509/000add53fca648e5bb7dbe81730bd38f.png
[Center 1]: /images/20220509/b842fe74200941ecbc615a0164d34b69.png
[Center 2]: /images/20220509/719edefecdab487b9f1d6f6af91fc6f6.png
[Center 3]: /images/20220509/782daa09b12b4ef4ae0b8a7d5d70e93e.png
[Center 4]: /images/20220509/bf146615e24642589908648a840d9ce6.png
[Center 5]: /images/20220509/5ed83cc461cb434094bfc81f8d5de0cc.png