Springboot 整合Mybatis-plus小测试

JS判断电脑(pc)还是移动端(mobile)

  返回  

RocketMQ 事务消息

2021/7/21 10:37:45 浏览:

事务

事务时恢复和并发控制的基本单位,简单来说,几个操作要么一起成功,要么一起失败,事务的4个特性: 原子性(Atomic) ,一致性 (Consistency) 隔离性 (Isolation) 持久性(Durablity) 也叫做 AICID 特性。

什么是本地事务

本地事务是指在单数据源上的访问和更新操作

transaction begin
insert/update/delete
...
transaction commit/rollback

一个事务只连接一个支持事务的数据源,需要保证事务的 ACID 特性,例如在下单常见,库存和订单如果在不同节点,就涉及分布式事务。

原理介绍

一些业务场景需要实现本地操作和消息发送事务一致性,即消息发送成功,则本地操作成功,消息发送失败, 本地操作失败(成功也需要回滚),以避免出现操作成功但消息发送失败,或者操作失败但消息发送成功的情况。
在这里插入图片描述

消息发送成功,事务操作成功时操作步骤如下所示:

  1. 生产者发送一条事务准备消息到事务消息队列。
  2. 生产者发送操作日志消息到操作日志队列,日志中包含步骤1消息的消息句柄。
  3. 生产者执行本地事务操作成功。
  4. 生产者请求修改消息延迟时间,使消息对消费者可见。
  5. 生产者向操作日志队列确认操作日志,删除日志消息。
  6. 消费者从事务消息队列中接收事务消息。
  7. 消费者处理事务消息。
  8. 消费者请求删除事务消息

消息发送成功,事务操作失败时操作步骤如下所示:

  1. 生产者发送一条事务准备消息到事务消息队列。
  2. 生产者发送操作日志信息到操作日志队列,日志中包含步骤1消息的消息句柄。
  3. 生产者执行本地事务操作失败。
  4. (对应上图中步骤A)操作日志队列向生产者发送消息,请求读取超时未确认操作日志
  5. (对应上图中步骤B)生产者检查事务结果,发现操作失败。
  6. (对应上图中步骤C)生产者提交回滚消息请求,不修改消息延迟时间,消息对消费者不可见。
  7. (对应上图中步骤D)生产者向操作日志队列确认操作日志,删除日志消息。

代码例子

public class TransactionMessageDemo {

    public class MyTransactionChecker implements TransactionChecker {
        @Override
        public boolean checkTransactionStatus(Message message) {
            boolean checkResult = false;
            String messageHandler = message.getReceiptHandle();
            try {
                //检查messageHandler事务是否成功。
                checkResult = true;
            } catch (Exception e) {
                checkResult = false;
            }
            return checkResult;
        }
    }

    public class MyTransactionOperations implements TransactionOperations {
        @Override
        public boolean doTransaction(Message message) {
            boolean transactionResult = false;
            String messageHandler = message.getReceiptHandle();
            String messageBody = message.getMessageBody();
            try {
                //根据messageHandler和messageBody执行本地事务。
                transactionResult = true;
            } catch (Exception e) {
                transactionResult = false;
            }
            return transactionResult;
        }
    }

    public static void main(String[] args) {
        System.out.println("Start TransactionMessageDemo");
        String transQueueName = "transQueueName";
        String accessKeyId = ServiceSettings.getMNSAccessKeyId();
        String accessKeySecret = ServiceSettings.getMNSAccessKeySecret();
        String endpoint = ServiceSettings.getMNSAccountEndpoint();

        CloudAccount account = new CloudAccount(accessKeyId, accessKeySecret, endpoint);
        MNSClient client = account.getMNSClient(); //初始化客户端。

        //为事务队列创建队列。
        QueueMeta queueMeta = new QueueMeta();
        queueMeta.setQueueName(transQueueName);
        queueMeta.setPollingWaitSeconds(15);

        TransactionMessageDemo demo = new TransactionMessageDemo();
        TransactionChecker transChecker = demo.new MyTransactionChecker();
        TransactionOperations transOperations = demo.new MyTransactionOperations();

        TransactionQueue transQueue = client.createTransQueue(queueMeta, transChecker);

        //执行事务。
        Message msg = new Message();
        String messageBody = "TransactionMessageDemo";
        msg.setMessageBody(messageBody);
        transQueue.sendTransMessage(msg, transOperations);

        //如果不使用队列,请删除队列并关闭客户端。
        transQueue.delete();

        //关闭客户端。
        client.close();
        System.out.println("End TransactionMessageDemo");
    }

}

RocketMQ 事务消息

事务模型
在这里插入图片描述

交互图

在这里插入图片描述

plantUML 代码

Producer -> SendMessageProcessor : sendTransactionMsg 发送事务消息
SenMessageProcessor -> TransactionMsgService: prepareMesage
TransactionMsgService -> TransactionBridge: putHalfMessage
TransactionBridge -> MessageStore: putMessage
MessageStore -> TransactionBridge : PutMsgResult
TransactionBride -> TransactionMsgService :putMsgResult
TransactionMsgService -> SendMessageProcessor :PutMsgResult
SendMessageProcessor -> Producer: PutMsgResult 
Producer -> Producer : executeTransaction
Producer -> EndTransactionProcessor : commit/rollback
EndTransactionProcessor -> TransactionService : commitMessage/rollbackMessage
TransactionService -> TransactionBridge :putOpMessage
TransactionBridge -> MessageStore : putMessage
MessageStore -> TransactionBridge :putMessageResult
TransactionBridge -> TransactionMsgService : putMsgResult
TransactionMsgService -> EndTransactionProcessor : PutMsgResult
EndTransactionProcessor -> EndTransactionProcessor : SendFinalMesssage
EndTransactionProcessor -> Consummer : ConsumeMessage 

时序图

在这里插入图片描述

事务型消息最终是为了保证最终一致性的,本地事务和消费者的操作保持一致。

  1. 首先,生产者会发送一个半成品消息发送给代理(不让消费者消费),此时代理返回给生产者一个 OK
  2. 本地执行任务,生产者访问 MySQL 执行一个事务
  3. 如果生产者本地事务执行成功了,生产者让代理告诉消费者可以消费消息了,如果生产者本地事务执行失败,生产者会告诉代理回滚,让消费者不要感知到消息。
  4. 生产者通知嗲了如果失败了,但是本地事务执行成功了,代理只收到一个半成品消息,但是久久收不到生产者的消息,这时需要一个回调机制去自动核查。
  5. 生产者收到核查通知后,检查数据是否提交成功
  6. 根据核查结果,将核查结果返回给代理(成功了 Commit, 失败了 rollback)

功能

  • 将半消息发送给代理
  • 生产者执行本地事务
  • 消费者回查生产者数据库

如果回查了几次,还是没有一个成功的反馈怎么办? 将消息放到死信队列中处理。

应用场景:异步扣减库存

在这里插入图片描述

  1. 创建一个流水,生产者再发消息给代理
  2. 生产者执行的本地事务是先扣减库存,再创建订单,最后需要更新流水状态和销量。
  3. 第2步成功了,消费者扣减数据库的库存,失败不扣减,保持一致
  4. 回查,消费者检查数据库中的流水。

欢迎关注:程序员开发者社区

在这里插入图片描述

参考资料
  • https://blog.csdn.net/weixin_44936189/article/details/116330353
  • http://rocketmq.apache.org/rocketmq/the-design-of-transactional-message/
  • https://help.aliyun.com/document_detail/34480.html
  • https://www.jianshu.com/p/04bce2b8b5ef

联系我们

如果您对我们的服务有兴趣,请及时和我们联系!

服务热线:18288888888
座机:18288888888
传真:
邮箱:888888@qq.com
地址:郑州市文化路红专路93号