RocketMQ进击

文章11 |   阅读 4692 |   点赞0

来源：https://blog.csdn.net/itanping/category_9355252.html

RocketMQ进击（六）磕一磕RocketMQ的事务消息和事务性消息的生产与消费

转x33g5p2x  于2021-12-19 转载在 其他  
字(10.5k)|赞(0)|评价(0)|浏览(262)

楔子：有句老话在电视上大概已经听得生茧：我们不成功便成仁。最终是要完成任务。

1. 不成功便成仁

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

Apache RocketMQ 在 4.3.0 版中已经支持分布式事务消息，它采用了 2PC 的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息。如下图所示：

上图说明了事务消息的大致方案，其中可以分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程。

一】事务消息发送及提交：

1）Producer 向 Broker 发送消息（half 消息）
2）服务端响应消息写入结果
3）Producer 根据发送结果执行本地事务（如果写入失败，此时 half 消息对业务不可见，本地逻辑不执行）
4）Producer 根据本地事务状态执行 Commit 或者 Rollback（Commit 操作生成消息索引，消息对消费者可见）

二】补偿流程：

5）对没有 Commit/Rollback 的事务消息（pending 状态的消息），从服务端发起一次“回查”
6）Producer 收到回查消息，检查回查消息对应的本地事务的状态
7）根据本地事务状态，重新 Commit 或者 Rollback

其中，补偿阶段用于解决消息 Commit 或者 Rollback 发生超时或者异常失败的情况。

2. 闲时二三闲事

2.1. 源码定义

2.1.1 事务的三种状态

事务消息共有三种状态，提交状态、回滚状态、中间状态：

  • TransactionStatus.CommitTransaction：提交事务，它允许消费者消费此消息。
  • TransactionStatus.RollbackTransaction：回滚事务，它代表该消息将被删除，不允许被消费。
  • TransactionStatus.Unknown：中间状态，它代表需要检查消息队列来确定状态。
package org.apache.rocketmq.client.producer;

public enum LocalTransactionState {
    COMMIT_MESSAGE,
    ROLLBACK_MESSAGE,
    UNKNOW;

    private LocalTransactionState() {
    }
}

2.2. 代码示例

这里通过一个常用场景简单模拟 RocketMQ 的事务消息：写2个微服务，分别是订单服务和商品服务。订单服务进行下单处理，并发送消息给商品服务，商品服务对下单成功的商品进行扣减库存。

2.2.1 创建事务性生产者

首先，写一个简易订单服务，使用 TransactionMQProducer 类创建生产者，并指定唯一的 ProducerGroup，就可以设置自定义线程池来处理这些检查请求。执行本地事务后、需要根据执行结果对消息队列进行状态回传。回传的三种事务状态如上所述。

package com.meiwei.service.mq.tcp.producer;

import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 订单服务
 */
public class OrderService {

    // Topic 为 Message 所属的一级分类，就像学校里面的初中、高中
    // Topic 名称长度不得超过 64 字符长度限制，否则会导致无法发送或者订阅
    private static final String MQ_CONFIG_TOPIC = "TOPIC_MEIWEI_SMS_NOTICE_TEST";
    // Tag 为 Message 所属的二级分类，比如初中可分为初一、初二、初三；高中可分为高一、高二、高三
    private static final String MQ_CONFIG_TAG_PUSH = "PID_MEIWEI_SMS_ORDER_TRANSACTION";

    public static void main(String[] args) throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer();

        // 指定 NameServer 地址列表，多个nameServer地址用半角分号隔开。此处应改为实际 NameServer 地址
        // NameServer 的地址必须有，但也可以通过启动参数指定、环境变量指定的方式设置，不一定要写死在代码里
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setProducerGroup("meiwei-producer-transaction-mq");

        // 自定义线程池，执行事务操作
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 20, 10L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(20), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("meiwei-order-service-transaction-msg-check");
                return thread;
            }
        });
        producer.setExecutorService(executor);

        // 设置事务消息监听器
        producer.setTransactionListener(new OrderTransactionListener());

        // 在发送MQ消息前，必须调用 start 方法来启动 Producer，只需调用一次即可
        producer.start();
        System.out.println("Order Server Start.");

        // 模拟业务
        for (int i = 0; i < 5; i++) {
            String orderId = System.currentTimeMillis() + "";
            String payOrder = "下单完成，订单编号：" + orderId;
            Message message = new Message(MQ_CONFIG_TOPIC, MQ_CONFIG_TAG_PUSH, orderId, payOrder.getBytes(RemotingHelper.DEFAULT_CHARSET));

            // 发送事务消息
            TransactionSendResult result = producer.sendMessageInTransaction(message, orderId);
            System.out.println("【发送事务消息】发送结果：" + result);

            Thread.sleep(100);
        }
    }
}

2.2.2 实现事务的监听接口

事务消息需要注册一个 TransactionListener，进行本地事务的执行和事务回查，代码如下：

package com.meiwei.service.mq.tcp.producer;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 订单事务消息监听器
 */
public class OrderTransactionListener implements TransactionListener {

    private static final Map<String, Integer> statusMap = new ConcurrentHashMap<>();

    // 执行本地事务
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        String orderId = (String) o;

        // 记录本地事务执行结果
        Integer status = this.executeTransactionResult(orderId);
        System.out.println("【订单事务消息监听器】本地有新订单，orderId: " + orderId + ", result: " + status);

        // 返回中间状态，需要检查消息队列来确定状态，即触发 checkLocalTransaction
        return LocalTransactionState.UNKNOW;
    }

    // 检查本地事务状态，并回应消息队列的检查请求
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        Integer status = statusMap.get(messageExt.getKeys());
        System.out.println("【订单事务消息监听器】执行事务消息回查，orderId: " + messageExt.getKeys() + ", result: " + status + "，时间: " + new Date());

        if (null != status) {
            switch (status) {
                case 0:
                    // 中间状态，它代表需要检查消息队列来确定状态
                    return LocalTransactionState.UNKNOW;
                case 1:
                    // 提交事务，它允许消费者消费此消息
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 2:
                    // 回滚事务，它代表该消息将被删除，不允许被消费
                    return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }

    // 模拟一个业务场景，并返回订单处理状态 
    private Integer executeTransactionResult(String orderId) {
        Integer status = Math.toIntExact(Long.valueOf(orderId) % 3);
        statusMap.put(orderId, status);
        return status;
    }
}

2.2.3 创建事务性消费者

再次，写一个简易的商品服务，接收订单服务的事务消息，如果消息成功 commit，则进行本地扣减库存。

package com.meiwei.service.mq.tcp.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;

/**
 * 商品服务
 */
public class DrugsService {

    // Topic 为 Message 所属的一级分类，就像学校里面的初中、高中
    // Topic 名称长度不得超过 64 字符长度限制，否则会导致无法发送或者订阅
    private static final String MQ_CONFIG_TOPIC = "TOPIC_MEIWEI_SMS_NOTICE_TEST";
    // Tag 为 Message 所属的二级分类，比如初中可分为初一、初二、初三；高中可分为高一、高二、高三
    private static final String MQ_CONFIG_TAG_PUSH = "PID_MEIWEI_SMS_ORDER_TRANSACTION";

    public static void main(String[] args) throws Exception {
        // 声明并初始化一个 consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
        // 指定 NameServer 地址列表，多个nameServer地址用半角分号隔开。此处应改为实际 NameServer 地址
        // NameServer 的地址必须有，但也可以通过启动参数指定、环境变量指定的方式设置，不一定要写死在代码里
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setConsumerGroup("meiwei-consumer-transaction-mq");
        // 设置 consumer 所订阅的 Topic 和 Tag，这里的 Topic 需要与生产者保持一致
        consumer.subscribe(MQ_CONFIG_TOPIC, MQ_CONFIG_TAG_PUSH);
        // 注册一个消息监听器消费消息
        consumer.registerMessageListener(new DrugsMqConsumerListener());

        consumer.start();
        System.out.println("Drugs Service Start.");
    }
}

2.2.4 实现消息消费监听器

package com.meiwei.service.mq.tcp.consumer;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Collections;
import java.util.List;
import java.util.Optional;

/**
 * 商品消费端监听器
 */
public class DrugsMqConsumerListener implements MessageListenerConcurrently {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        Optional.ofNullable(list).orElse(Collections.emptyList()).forEach(msg -> {
            String orderId = msg.getKeys();
            System.out.println("【商品消费端监听器】您有新订单，orderId: " + orderId + "，商品库存需要更新");
        });

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

2.3. 测试与结果

订单服务作为生产者发出新订单扣减库存消息：

Order Server Start.
【订单事务消息监听器】本地有新订单，orderId: 1571021207194, result: 1
【发送事务消息】发送结果：SendResult [sendStatus=SEND_OK, msgId=0A06341B156C18B4AAC24542D3BF0000, offsetMsgId=null, messageQueue=MessageQueue [topic=TOPIC_MEIWEI_SMS_NOTICE_TEST, brokerName=YYW-SH-PC-1454, queueId=1], queueOffset=1620]
【订单事务消息监听器】本地有新订单，orderId: 1571021207595, result: 0
【发送事务消息】发送结果：SendResult [sendStatus=SEND_OK, msgId=0A06341B156C18B4AAC24542D42B0001, offsetMsgId=null, messageQueue=MessageQueue [topic=TOPIC_MEIWEI_SMS_NOTICE_TEST, brokerName=YYW-SH-PC-1454, queueId=2], queueOffset=1621]
【订单事务消息监听器】本地有新订单，orderId: 1571021207696, result: 2
【发送事务消息】发送结果：SendResult [sendStatus=SEND_OK, msgId=0A06341B156C18B4AAC24542D4900002, offsetMsgId=null, messageQueue=MessageQueue [topic=TOPIC_MEIWEI_SMS_NOTICE_TEST, brokerName=YYW-SH-PC-1454, queueId=3], queueOffset=1622]
【订单事务消息监听器】本地有新订单，orderId: 1571021207798, result: 2
【发送事务消息】发送结果：SendResult [sendStatus=SEND_OK, msgId=0A06341B156C18B4AAC24542D4F60003, offsetMsgId=null, messageQueue=MessageQueue [topic=TOPIC_MEIWEI_SMS_NOTICE_TEST, brokerName=YYW-SH-PC-1454, queueId=0], queueOffset=1623]
【订单事务消息监听器】本地有新订单，orderId: 1571021207900, result: 2
【发送事务消息】发送结果：SendResult [sendStatus=SEND_OK, msgId=0A06341B156C18B4AAC24542D55C0004, offsetMsgId=null, messageQueue=MessageQueue [topic=TOPIC_MEIWEI_SMS_NOTICE_TEST, brokerName=YYW-SH-PC-1454, queueId=1], queueOffset=1624]
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207194, result: 1，时间: Mon Oct 14 10:46:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:46:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207798, result: 2，时间: Mon Oct 14 10:46:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207696, result: 2，时间: Mon Oct 14 10:46:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207900, result: 2，时间: Mon Oct 14 10:46:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:47:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:48:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:49:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:50:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:51:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:52:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:53:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:54:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:55:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:56:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:57:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:58:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 10:59:56 CST 2019
【订单事务消息监听器】执行事务消息回查，orderId: 1571021207595, result: 0，时间: Mon Oct 14 11:00:56 CST 2019

商品服务作为消费者监听并消费到扣减库存的消息，后面就可以进行扣减库存的业务处理：

从上面的生产者输出日志可以看到订单服务发出的 5 条消息：

  • 订单状态为 UNKNOW：0 的有 1 条，这条消息被回查了 15 次（transactionCheckMax 默认值 15，可配置）
  • 订单状态为 COMMIT_MESSAGE：1 的有 1 条，且被商品服务正常消费
  • 订单状态为 ROLLBACK_MESSAGE：2 的有 3 条，且未被商品服务消费，也未被回查

2.4. 使用限制

  1. 事务消息不支持延时消息和批量消息。
  2. 事务性消息可能不止一次被检查或消费。
  3. 为了避免单个消息被检查太多次而导致半队列消息累积，源码默认将单个消息的检查次数限制为 15 次，但是我们可以通过 Broker 配置文件的 transactionCheckMax 参数来修改这个限制。如果已经检查某条消息超过 N 次的话（N =  transactionCheckMax），Broker 则将丢弃此消息，并在默认情况下同时打印错误日志。不过可以通过重写  AbstractTransactionCheckListener 类来修改这个行为。
  4. 事务消息 将在 Broker 配置文件中的参数 transactionMsgTimeout 这样的特定时间长度之后 被检查。不过可以通过设置属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于 transactionMsgTimeout 参数。
  5. 提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。
  6. 事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/itanping/article/details/101372336

内容来源于网络，如有侵权，请联系作者删除！

Javarocketmqmq事务消息

相关文章

    查看更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com