【RocketMQ】事务实现原理

rocketmq,事务,实现,原理 · 浏览次数 : 214

小编点评

**生成内容** **1. 确定事务状态** * 如果发送成功,则设置事务状态为COMMIT_MESSAGE。 * 如果回滚事务,则设置事务状态为ROLLBACK_MESSAGE。 * 如果未知状态,则设置事务状态为UNKNOW。 **2. 执行本地事务** * 如果事务状态为COMMIT_MESSAGE,则执行本地事务并返回结果。 * 如果事务状态为ROLLBACK_MESSAGE,则回滚本地事务并返回结果。 * 如果事务状态为UNKNOW,则不做任何操作。 **3. 检查事务结果** * 在事务状态检查时,获取半主题的消费进度。 * 从半主题中拉取消息,并将其放入到移除集合中。 * 遍历移除集合,删除已删除的half消息。 **4. 处理半消息** * 从OP队列中拉取消息,并将其放入到 removeMap中。 * 检查removeMap中是否包含当前消息。 * 如果消息已删除,则将其从OP队列中移除。 * 如果消息未删除,则将其重新加入OP队列中。

正文

RocketMQ事务的使用场景

单体架构下的事务

在单体系统的开发过程中,假如某个场景下需要对数据库的多张表进行操作,为了保证数据的一致性,一般会使用事务,将所有的操作全部提交或者在出错的时候全部回滚。以创建订单为例,假设下单后需要做两个操作:

  1. 在订单表生成订单
  2. 在积分表增加本次订单增加的积分记录

在单体架构下只需使用@Transactional开启事务,就可以保证数据的一致性:

    @Transactional
    public void order() {
        String orderId = UUID.randomUUID().toString();
        // 生成订单
        orderService.createOrder(orderId);
        // 增加积分
        creditService.addCredits(orderId);
    }

然而现在越来越多系统开始使用分布式架构,在分布式架构下,订单系统和积分系统可能是两个独立的服务,此时就不能使用上述的方法开启事务了,因为它们不处于同一个事务中,在出错的情况下,无法进行全部回滚,只能对当前服务的事务进行回滚,所以就有可能出现订单生成成功但是积分服务增加积分失败的情况(也可能相反),此时数据处于不一致的状态。

分布式架构下的事务

分布式架构下如果需要保证事务的一致性,需要使用分布式事务,分布式事务的实现方式有多种,这里我们先看通过RocketMQ事务的实现方式。

同样以下单流程为例,在分布式架构下的处理流程如下:

  1. 订单服务生成订单
  2. 发送订单生成的MQ消息,积分服务订阅消息,有新的订单生成之后消费消息,增加对应的积分记录

普通MQ消息存在的问题

如果使用@Transactional + 发送普通MQ的方式,看下存在的问题:

  1. 假如订单创建成功,MQ消息发送成功,但是order方法在返回的前一刻,服务突然宕机,由于开启了事务,事务还未提交(方法结束后才会正常提交),所以订单表并未生成记录,但是MQ却已经发送成功并且被积分服务消费,此时就会存在订单未创建但是积分记录增加的情况
  2. 假如先发送MQ消息再创建订单呢,此时问题就更明显了,如果MQ消息发送成功,创建订单失败,那么同样处于不一致的状态
    @Transactional
    public void order() {
        String orderId = UUID.randomUUID().toString();
        // 创建订单
        Order order = orderService.createOrder(orderDTO.getOrderId());
        // 发送订单创建的MQ消息
        sendOrderMessge(order);
        return;
    }

解决上述问题的方式就是使用RocketMQ事务消息。

RocketMQ事务消息的使用

使用事务消息需要实现自定义的事务监听器,TransactionListener提供了本地事务执行和状态回查的接口,executeLocalTransaction方法用于执行我们的本地事务,checkLocalTransaction是一种补偿机制,在异常情况下如果未收到事务的提交请求,会调用此方法进行事务状态查询,以此决定是否将事务进行提交/回滚:

public interface TransactionListener {
    /**
     * 执行本地事务
     *
     * @param msg Half(prepare) message half消息
     * @param arg Custom business parameter
     * @return Transaction state
     */
    LocalTransactionState executeLocalTransaction(final Message msg, final Object arg);

    /**
     * 本地事务状态回查
     *
     * @param msg Check message
     * @return Transaction state
     */
    LocalTransactionState checkLocalTransaction(final MessageExt msg);
}

这里我们实现自定义的事务监听器OrderTransactionListenerImpl:

  • executeLocalTransaction方法中创建订单,如果创建成功返回COMMIT_MESSAGE,如果出现异常返回ROLLBACK_MESSAGE
  • checkLocalTransaction方法中回查事务状态,根据消息体中的订单ID查询订单是否已经创建,如果创建成功提交事务,如果未获取到认为失败,此时回滚事务。
public class OrderTransactionListenerImpl implements TransactionListener {

    @Autowired
    private OrderService orderService;

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            String body = new String(msg.getBody(), Charset.forName("UTF-8"));
            OrderDTO orderDTO = JSON.parseObject(body, OrderDTO.class);
            // 模拟生成订单
            orderService.createOrder(orderDTO.getOrderId());
        } catch (Exception e) {
            // 出现异常,返回回滚状态
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        // 创建成功,返回提交状态
        return LocalTransactionState.COMMIT_MESSAGE;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        String body = new String(msg.getBody(), Charset.forName("UTF-8"));
        OrderDTO orderDTO = JSON.parseObject(body, OrderDTO.class);
        try {
            // 根据订单ID查询订单是否存在
            Order order = orderService.getOrderByOrderId(orderDTO.getOrderId());
            if (null != order) {
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
}

接下来看如何发送事务消息,事务消息对应的生产者为TransactionMQProducer,创建TransactionMQProducer之后,设置上一步自定义的事务监听器OrderTransactionListenerImpl,然后将订单ID放入消息体中, 调用sendMessageInTransaction发送事务消息:

public class TransactionProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        // 创建下单事务监听器
        TransactionListener transactionListener = new OrderTransactionListenerImpl();
        // 创建生产者
        TransactionMQProducer producer = new TransactionMQProducer("order_group");
        // 事务状态回查线程池
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });
        // 设置线程池
        producer.setExecutorService(executorService);
        // 设置事务监听器
        producer.setTransactionListener(transactionListener);
        // 启动生产者
        producer.start();
        try {
            // 创建订单消息
            OrderDTO orderDTO = new OrderDTO();
            // 模拟生成订单唯一标识
            orderDTO.setOrderId(UUID.randomUUID().toString());
            // 转为字节数组
            byte[] msgBody = JSON.toJSONString(orderDTO).getBytes(RemotingHelper.DEFAULT_CHARSET);
            // 构建消息
            Message msg = new Message("ORDER_TOPIC", msgBody);
            // 调用sendMessageInTransaction发送事务消息
            SendResult sendResult = producer.sendMessageInTransaction(msg, null);
            System.out.printf(sendResult.toString());
            Thread.sleep(10);
        } catch (MQClientException | UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }
        producer.shutdown();
    }
}

事务的执行流程:

  1. 在订单服务下单后,向Borker发送生成订单的事务消息,投递到ORDER_TOPIC主题中
  2. Broker收到事务消息之后,不会直接投递到ORDER_TOPIC主题中,而是先放在另外一个主题中,也叫half主题,half主题对消费者不可见
  3. half主题加入消息成功之后,会回调事务监听器的的executeLocalTransaction方法,执行本地事务,也就是订单创建,如果创建成功返回COMMIT状态,如果出现异常返回ROLLBACK状态
  4. 根据上一步的返回状态,进行结束事务的处理
    • 提交:从half主题中删除消息,然后将消息投送到ORDER_TOPIC主题中,积分服务订阅ORDER_TOPIC主题进行消费,生成积分记录
    • 回滚:从half主题中删除消息即可
  5. 如果本地事务返回的执行结果状态由于网络原因或者其他原因未能成功的发送给Broker,Broker未收到事务的执行结果,在补偿机制定时检查half主题中消息的事务执行状态时,会回调事务监听器checkLocalTransaction的接口,进行状态回查,判断订单是否创建成功,然后进行结束事务的处理

使用事务消息不会存在订单创建失败但是消息发送成功的情况,不过你可能还有一个疑问,假如订单创建成功了,消息已经投送到队列中,但是积分服务在消费的时候失败了,这样数据还是处于不一致的状态,个人感觉,积分服务可以在失败的时候进行重试或者进行一些其他的补偿机制来保证积分记录成功的生成,在极端情况下积分记录依旧没有生成,此时可能就要人工接入处理了。

RocketMQ事务实现原理

事务消息发送

一、 生产者发送事务消息
生产者在发送事务消息的时候,会在消息属性中设置PROPERTY_TRANSACTION_PREPARED属性,然后向Broker发送消息。

Broker收到消息后,会判断消息是否含有PROPERTY_TRANSACTION_PREPARED属性,如果没有该属性,表示是普通消息,按照普通消息的写入流程执行即可,如果有该属性
表示开启事务,还不能直接加入到实际的消息队列中,否则一旦加入就会被消费者消费,所以需要先对消息暂存,等收到消息提交请求时才可以添加到实际的消息队列中,RocketMQ设置了一个RMQ_SYS_TRANS_HALF_TOPIC主题来暂存事务消息,放入这个主题中的消息被称为half消息,它的处理逻辑如下:

  1. 设置消息实际的主题和队列ID,待收到事务提交请求后恢复实际的主题和队列ID,向实际的队列中添加消息(类似于延迟消息的实现);
  2. 更改消息的主题为half消息主题RMQ_SYS_TRANS_HALF_TOPIC
  3. 更改消息队列,默认使用RMQ_SYS_TRANS_HALF_TOPIC主题下ID为0的那个消息队列,将消息先投递到此队列中;

Broker写入half消息之后,会向生产者返回half消息的写入结果状态。

二、 执行本地事务
在上一步中,生产者向Broker发送了事务消息,生产者会根据返回的响应结果状态来判断消息是否发送成功:
(1)发送成功,此时执行本地事务,并返回本地事务执行结果状态,执行结果一般有以下三种;
* COMMIT_MESSAGE:表示执行成功;
* ROLLBACK_MESSAGE:执行失败需要回滚事务;
* UNKNOW:未知状态;
(2)未发送成功,比如FLUSH_DISK_TIMEOUT刷盘超时、FLUSH_SLAVE_TIMEOUTSLAVE_NOT_AVAILABLE从节点不可用等状态,此时意味着消息发送失败,本地事务状态置为ROLLBACK_MESSAGE准备回滚事务;

三、结束事务

经过了前两步骤之后,消息暂存在Broker的half主题中,也得到了本地事务的执行结果状态,接下来就需要根据本地事务的执行结果状态来决定回滚还是提交事务,首先会构建一个结束事务的请求头EndTransactionRequestHeader,请求头中会设置消息的偏移量等信息,然后根据事务的执行结果来设置不同的标识,上面知道事务执行结果一般有三种状态:

  1. COMMIT_MESSAGE:表示执行成功,可以提交事务,请求头中设置TRANSACTION_COMMIT_TYPE标识表示提交事务;
  2. ROLLBACK_MESSAGE:表示需要回滚事务,请求头中设置TRANSACTION_ROLLBACK_TYPE标识进行事务回滚;
  3. UNKNOW:事务执行结果未知状态,请求头中设置TRANSACTION_NOT_TYPE标识未知状态的事务;

之后会向Broker发送这个结束事务的请求,Broker收到请求后会做如下处理:

  1. 判断自己是否是从节点,从节点没有结束事务的权限,如果是从节点返回SLAVE_NOT_AVAILABLE状态;
  2. 从请求头中获取事务的提交类型,如果是TRANSACTION_NOT_TYPE打印warn信息,然后返回NULL,如果是其他类型,做如下处理:
    (1)TRANSACTION_COMMIT_TYPE标识:表示提交事务,请求信息中携带了消息的偏移量,会根据偏移量先查找消息是否存在,如果存在与请求头中携带的消息信息进行对比校验是否一致,校验通过才可以提交事务,此时会恢复消息原本的主题和队列,将消息投递到对应的队列中,然后将对应的half消息进行删除;
    (2)TRANSACTION_ROLLBACK_TYPE标识:表示回滚事务,同样会先根据请求中的消息偏移量进行查找并校验,通过之后,将对应的half消息进行删除;

消息删除

由于CommitLog追加写的性质,RocketMQ并不会直接将half消息从CommitLog中删除,而是使用了另外一个主题RMQ_SYS_TRANS_OP_HALF_TOPIC以下简称OP主题/队列),将已经删除的half消息记录在OP主题队列中,在事务状态检查时,需要通过这个OP队列来判断消息是否被标记了删除。

事务状态检查

由于各种原因有可能未成功收到提交/回滚事务的请求,所以RocketMQ需要定期检查half消息,检查事务的执行结果。在检查的时候会获取half主题(RMQ_SYS_TRANS_HALF_TOPIC)下的所有消息队列,遍历所有的half消息队列,对队列中的消息进行处理。

每个half消息队列,会有一个对应的OP队列,里面记录了被删除的half消息,首先需要从这个OP队列中拉取消息(因为不知道每条消息在OP队列中的哪个位置,所以需要不断拉取进行查找,每次会拉取32条),并放入到一个集合removeMap中,用于判断当前消息是否已经被标记了删除。

Broker记录了每个half队列的消费进度,每次检查时会获取上一次处理的位置,从这个位置之后继续处理队列中的每一条消息:

  1. 时间校验,如果当前时间减去检查开始时间大于最大处理时间,表示此次检查超时,终止循环等待下一次;
  2. 如果removeMap中包含当前消息,表示消息已经被删除,不需要进行处理;
  3. 如果removeMap不包含当前half消息,会根据消息偏移量获取half消息,如果消息获取不为空继续下一步;
  4. 判断当前half消费是否需要被丢弃或者跳过:
    • 丢弃:每次检查会记录本条消息的检查次数,并记在属性中,如果超过了最大的次数,表示消息需要做丢弃处理;
    • 跳过:如果消息在队列中的存留时间是否超过了设置的最大的保留时间,表示需要跳过,不进行处理;
  5. 判断消息的的存入时间是否大于本次开始检查的时间,如果大于说明是新加入的消息,由于事务消息发送后不会立刻提交,所以此时暂不需要进行检查,中断当前处理等待下一次检查;
  6. 如果消息属性中设置了PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS(事务最晚回查时间),判断half消息的存留时间是否超过了个值,如果未超过说明此时还未到回查的时间,并且当前消息未被删除,会将当前的消息重新加入half队列中,因为需要继续往后处理并在结束时更新进度,如果不重新将消息加入到队列中,这条消息就没办法再次处理;
  7. 判断是否需要进行事务回查,发送回查请求(处理回查请求时会调用checkLocalTransaction进行状态检查),回查请求通过线程池异步实现的,所以需要将half消息重新加入到队列中等待下次检查;
  8. 更新half队列的处理进度和OP队列的消费进度;

事务相关源码:【RocketMQ】【源码】事务的实现原理

参考

RocketMQ事务官方文档

与【RocketMQ】事务实现原理相似的内容:

【RocketMQ】事务实现原理

RocketMQ事务的使用场景 单体架构下的事务 在单体系统的开发过程中,假如某个场景下需要对数据库的多张表进行操作,为了保证数据的一致性,一般会使用事务,将所有的操作全部提交或者在出错的时候全部回滚。以创建订单为例,假设下单后需要做两个操作: 在订单表生成订单 在积分表增加本次订单增加的积分记录

记一次RocketMQ消费非顺序消息引起的线上事故

应用场景 C端用户提交工单、工单创建完成之后、会发布一条工单创建完成的消息事件(异步消息)、MQ消费者收到消息之后、会通知各处理器处理该消息、各处理器处理完后都会发布一条将该工单写入搜索引擎的消息、最终该工单出现在搜索引擎、被工单处理人检索和处理。 事故异常体现 1、异常体现 从工单的流转记录发现、

【主流技术】聊一聊消息队列 RocketMQ 的基本结构与概念

RocketMQ 是阿里巴巴在 2012 年开源的分布式消息中间件,目前已经捐赠给 Apache 软件基金会,并于 2017 年 9 月 25 日成为 Apache 的顶级项目。 作为经历过多次阿里巴巴双十一这种“超级工程”的洗礼并有稳定出色表现的国产中间件,以其高性能、低延时和高可靠等特性近年来...

RocketMQ 事件驱动:云时代的事件驱动有啥不同?

本文深入探讨了云时代 EDA 的新内涵及它在云时代再次流行的主要驱动力,包括技术驱动力和商业驱动力,随后重点介绍了 RocketMQ 5.0 推出的子产品 EventBridge,并通过几个云时代事件驱动的典型案例,进一步叙述了云时代事件驱动的常见场景和最佳实践。

RocketMq开启安全认证ACL-解决服务器系统安全漏洞

1、为什么要开启ACL 通过之前的文章我们已经知道怎么安装RocketMq了。如果你还不会安装RocketMq可以查看我的这篇文章:快速入门一篇搞定RocketMq-实现微服务实战落地 进行软件安装,附文章地址:https://www.cnblogs.com/sowler/p/18173752 。虽

如何实现一个简单易用的 RocketMQ SDK

2018 年,做为架构负责人,接到一个架构需求:实现一个简单易用的 RocketMQ SDK 。 因为各个团队 RocketMQ 原生客户端配置起来千奇百怪,有的配置存在风险,各团队负责人都需要一个简洁易用的 RocketMQ SDK 。 我立马调研相关开源的方案,当时 RocketMQ-Sprin

RocketMQ 之 IoT 消息解析:物联网需要什么样的消息技术?

前言: 从初代开源消息队列崛起,到 PC 互联网、移动互联网爆发式发展,再到如今 IoT、云计算、云原生引领了新的技术趋势,消息中间件的发展已经走过了 30 多个年头。 目前,消息中间件在国内许多行业的关键应用中扮演着至关重要的角色。随着数字化转型的深入,客户在使用消息技术的过程中往往同时涉及交叉场

[转帖]RocketMQ - nameSrv和Broker

RocketMQ RocketMQ是一个统一的消息传递引擎,轻量级的数据处理平台。 Name Server Name Server充当路由消息的提供者,生产者(Producer)或消费者(Customer)可以通过Name Server查找各主题对应的Broker IP列表,多个Name Serve

RocketMQ - 生产者原理

https://rocketmq.apache.org/ Apache RocketMQ是一款开源的、分布式的消息投递与流数据平台。出生自阿里巴巴,在阿里巴巴内部经历了3个版本后,作为Apache 顶级开源项目之一直到现在。在GitHub上有10000+star、5000+fork、170+cont

RocketMQ - 生产者启动流程

生产者启动流程 DefaultMQProducer是RocketMQ中默认的生产者实现 核心属性: namesrvAddr: 继承自 ClientConfig,表示 RocketMQ 集群的Namesrv 地址,如果是多个则用分号分开。比如:127.0.0.1:9876;127.0.0.2:9876