如何用 RabbitMQ 解决分布式事务？

傷城~ 2023-10-01 22:25 22阅读 0赞

要说哪种分布式事务处理方案效率高，必然绕不开消息中间件！基于消息中间件的两阶段提交方案，通常用在高并发场景下。这种方式通过牺牲数据的强一致性换取性能的大幅提升，不过实现这种方式的成本和复杂度是比较高的，使用时还要看实际业务情况。

今天小编想通过一个简单的案例，来和大家聊一聊如何通过消息中间件来处理分布式事务。

## 1. 思路分析 ##

先来说说整体思路。

有一个名词叫做消息驱动的微服务，相信很多小伙伴都听说过。怎么理解呢？

在微服务系统中，服务之间的互相调用，我们可以使用 HTTP 的方式，例如 OpenFeign，也可以使用 RPC 的方式，例如 Dubbo，除了这些方案之外，我们也可以使用消息驱动，这是一种典型的响应式系统设计方案。

在消息驱动的微服务中，服务之间不再互相直接调用，当服务之间需要通信时，就把通信内容发送到消息中间件上，另一个服务则通过监听消息中间件中的消息队列，来完成相应的业务逻辑调用，过程就是这么个过程，并不难，具体怎么玩，我们继续往下看。

## 2. 业务分析 ##

折腾了半天，后来小编在网上找到了一个别人写好的例子，我觉得用来演示这个问题特别合适，所以我就没有自己写案例了，直接用别人的代码，我们来逐个分析，跟前面讲分布式事务 Seata 的方式一致。

首先我们来看如下一张流程图，这是一个用户购票的案例：

![1b6198bee0d72ab91f89cc294fdbf6c0.png][]

当用户想要购买一张票时：

1.  向新订单队列中写入一条数据。
2.  Order Service 负责消费这个队列中的消息，完成订单的创建，然后再向新订单缴费队列中写入一条消息。
3.  User Service 负责消费新订单缴费队列中的消息，在 User Service 中完成对用户账户余额的划扣，然后向新订单转移票队列中写入一条消息。
4.  Ticket Service 负责消费新订单转移票队列，在 Ticket Service 中完成票的转移，然后发送一条消息给订单完成队列。
5.  最后 Order Service 中负责监听订单完成队列，处理完成后的订单。

这就是一个典型的**消息驱动微服务**，也是一个典型的**响应式系统**。在这个系统中，一共有三个服务，分别是：

*  Order Service
 *  User Service
 *  Ticket Service

这三个服务之间不会进行任何形式的直接调用，大家有事都是直接发送到消息中间件，其他服务则从消息中间件中获取自己想要的消息然后进行处理。

具体到我们的实践中，则多了一个检查票是否够用的流程，如下图：

![facfd332bef94aaeb4bb09d180e2cb45.png][]

创建订单时，先由 Ticket 服务检查票是否够用，没问题的话再继续发起订单的创建。其他过程我就不说了。

**另外还需要注意，在售票系统中，由于每张票都不同，例如每张票可能有座位啥的，因此一张票在数据库中往往是被设计成一条记录。**

## 3. 实践 ##

流程我已经说明白了，接下来我们就来看看具体的代码实践。

## 3.1 准备数据库 ##

首先我们准备三个数据库，分别是：

*  javaboy\_order：订单库，用户创建订单等操作，在这个数据库中完成。
 *  javaboy\_ticket：票务库，这个库中保存着所有的票据信息，每一张票都是一条记录，都保存在这个库中。
 *  javaboy\_user：用户库，这里保存着用户的账户余额以及付款记录等信息。

每个库中都有各自对应的表，为了操作方便，这些表不用自己创建，将来等项目启动了，利用 JPA 自动创建即可。

## 3.2 项目概览 ##

我们先来整体上看下这个项目，私信小编【111】可以下载完整代码：

![62760db00deb5beef9522aabf1c9eb56.png][]

一共有五个服务：

*  eureka：注册中心
 *  order：订单服务
 *  service：公共模块
 *  ticket：票务服务
 *  user：用户服务

下面分别来说。

## 3.3 注册中心 ##

有人说，都消息驱动了，还要注册中心干嘛？

消息驱动没错，消息驱动微服务之后每个服务只管把消息往消息中间件上扔，每个服务又只管消费消息中间件上的消息，这个时候对于服务注册中心似乎不是那么强需要。不过在我们这个案例中，消息驱动主要用来处理事务问题，其他常规需求我们还是用 OpenFeign 来处理，所以这里我们依然需要一个注册中心。

这里的注册中心我就选择常见的 Eureka，省事一些。**由于本文主要是和大家聊分布式事务，所以涉及到微服务的东西我就简单介绍下，不会占用过多篇幅**

服务注册中心的创建记得加上 Spring Security，将自己的服务注册中心保护起来。

这块有一个小小的细节和大家多说两句。

Eureka 用 Spring Security 保护起来之后，以后其他服务注册都是通过 Http Basic 来认证，所以我们要在代码中开启 Http Basic 认证，如下（以前旧版本不需要下面这段代码，但是新版本需要）：

@Configuration
    public class SecurityConfig extends WebSecurityConfigurerAdapter {
        @Override
        protected void configure(HttpSecurity http) throws Exception {
            http.authorizeRequests()
                    .anyRequest().authenticated()
                    .and()
                    .httpBasic()
                    .and().formLogin().and().csrf().disable();
        }
    }

## 3.4 购票服务 ##

接下来我们就来看看购票服务。

购票是从下订单开始，所以我们就先从订单服务 order 开始整个流程的分析。

## 3.4.1 新订单处理（order） ##

当用户发起一个购票请求后，这个请求发送到 order 服务上，order 服务首先会向 order:new 队列发送一条消息，开启一个订单的处理流程。代码如下：

@Transactional
    @PostMapping("")
    public void create(@RequestBody OrderDTO dto) {
        dto.setUuid(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("order:new", dto);
    }

上面设置的 UUID 是整个订单在处理过程中的一个唯一标志符，也算是一条主线。

order:new 队列中的消息将被 ticket 服务消费，ticket 服务消费order:new 中的消息，并进行锁票操作（锁票的目的防止有两个消费同时购买同一张票），锁票成功后，ticket 服务将向 order:locked 队列发送一条消息，表示锁票成功；否则向 order:fail 队列发送一条消息表示锁票失败。

这里的 OrderDTO 对象将贯穿整个购票过程。

## 3.4.2 锁票（ticket） ##

锁票操作是在 ticket 服务中完成的，代码如下：

@Transactional
    @RabbitListener(queues = "order:new")
    public void handleTicketLock(OrderDTO msg) {
        LOG.info("Get new order for ticket lock:{}", msg);
        int lockCount = ticketRepository.lockTicket(msg.getCustomerId(), msg.getTicketNum());
        if (lockCount == 0) {
            msg.setStatus("TICKET_LOCK_FAIL");
            rabbitTemplate.convertAndSend("order:fail", msg);
        } else {
            msg.setStatus("TICKET_LOCKED");
            rabbitTemplate.convertAndSend("order:locked", msg);
        }
    }

先调用 lockTicket 方法去数据库中锁票，所谓的锁票就是将要购买的票的 lock\_user 字段设置为 customer\_id（购买者的 id）。

如果锁票成功（即数据库修改成功），设置 msg 的状态为 TICKET\_LOCKED，同时发送消息到 order:locked 队列，表示锁票成功。

如果锁票失败（即数据库修改失败），设置 msg 的状态为 TICKET\_LOCK\_FAIL，同时发送消息到 order:fail 队列，表示锁票失败。

## 3.4.2 锁票成功（order） ##

接下来，由 order 服务消费 order:locked 队列中的消息，也就是锁票成功后接下来的操作。

@Transactional
    @RabbitListener(queues = "order:locked")
    public void handle(OrderDTO msg) {
        LOG.info("Get new order to create:{}", msg);
        if (orderRepository.findOneByUuid(msg.getUuid()) != null) {
            LOG.info("Msg already processed:{}", msg);
        } else {
            Order order = newOrder(msg);
            orderRepository.save(order);
            msg.setId(order.getId());
        }
        msg.setStatus("NEW");
        rabbitTemplate.convertAndSend("order:pay", msg);
    }

锁票成功后，先根据订单的 UUID 去订单数据库查询，是否已经有订单记录了，如果有，说明这条消息已经被处理了，可以防止订单的重复处理（这块主要是解决幂等性问题）。

如果订单还没有被处理，则创建一个新的订单对象，并保存到数据库中，创建新订单对象的时候，需要设置订单的 status 为 NEW。

最后设置 msg 的 status 为 NEW，然后向 order:pay 队列发送一条消息开启付款流程，付款是由 user 服务提供的。user 服务中会检查用户的账户余额是否够用，如果不够用，就会发送消息到 order:ticket\_error 队列，表示订票失败；如果余额够用，则进行正常的付款操作，并在付款成功后发送消息到 order:ticket\_move 队列，开启票的转移。

## 3.4.3 缴费（user） ##

锁票成功后，接下来就是付费了，付费服务由 user 提供。

@Transactional
    @RabbitListener(queues = "order:pay")
    public void handle(OrderDTO msg) {
        LOG.info("Get new order to pay:{}", msg);
        // 先检查payInfo判断重复消息。
        PayInfo pay = payInfoRepository.findOneByOrderId(msg.getId());
        if (pay != null) {
            LOG.warn("Order already paid, duplicated message.");
            return;
        }
        Customer customer = customerRepository.getById(msg.getCustomerId());
        if (customer.getDeposit() < msg.getAmount()) {
            LOG.info("No enough deposit, need amount:{}", msg.getAmount());
            msg.setStatus("NOT_ENOUGH_DEPOSIT");
            rabbitTemplate.convertAndSend("order:ticket_error", msg);
            return;
        }
        pay = new PayInfo();
        pay.setOrderId(msg.getId());
        pay.setAmount(msg.getAmount());
        pay.setStatus("PAID");
        payInfoRepository.save(pay);
        customerRepository.charge(msg.getCustomerId(), msg.getAmount());
        msg.setStatus("PAID");
        rabbitTemplate.convertAndSend("order:ticket_move", msg);
    }

这里的执行步骤如下：

1.  首先根据订单 id 去查找付款信息，检查当前订单是否已经完成付款，如果已经完成服务，则直接 return，这一步也是为了处理幂等性问题。
2.  根据顾客的 id，查找到顾客的完整信息，包括顾客的账户余额。
3.  检查顾客的账户余额是否足够支付票价，如果不够，则设置 msg 的 status 为 NOT\_ENOUGH\_DEPOSIT，同时向 order:ticket\_error 队列发送消息，表示订票失败。
4.  如果顾客账户余额足够支付票价，则创建一个 PayInfo 对象，设置相关的支付信息，并存入 pay\_info 表中。
5.  调用 charge 方法完成顾客账户余额的扣款。
6.  发送消息到 order:ticket\_move 队列中，开启交票操作。

## 3.4.4 交票（ticket） ##

@Transactional
    @RabbitListener(queues = "order:ticket_move")
    public void handleTicketMove(OrderDTO msg) {
        LOG.info("Get new order for ticket move:{}", msg);
        int moveCount = ticketRepository.moveTicket(msg.getCustomerId(), msg.getTicketNum());
        if (moveCount == 0) {
            LOG.info("Ticket already transferred.");
        }
        msg.setStatus("TICKET_MOVED");
        rabbitTemplate.convertAndSend("order:finish", msg);
    }

调用 moveTicket 方法完成交票操作，也就是设置 ticket 表中票的 owner 为 customerId。

交票成功后，发送消息到 order:finish 队列，表示交票完成。

## 3.4.5 订单完成（order） ##

@Transactional
    @RabbitListener(queues = "order:finish")
    public void handleFinish(OrderDTO msg) {
        LOG.info("Get finished order:{}", msg);
        Order order = orderRepository.getById(msg.getId());
        order.setStatus("FINISH");
        orderRepository.save(order);
    }

这里的处理就比较简单，订单完成后，就设置订单的状态为 FINISH 即可。

上面介绍的是一条主线，顺利的话，消息顺着这条线走一遍，一个订单就处理完成了。

不顺利的话，就有各种幺蛾子，我们分别来看。

## 3.4.6 锁票失败（order） ##

锁票是在 ticket 服务中完成的，如果锁票失败，就会直接向 order:fail 队列发送消息，该队列的消息由 order 服务负责消费。

## 3.4.7 扣款失败（ticket） ##

扣款操作是在 user 中完成的，扣款失败就会向 order:ticket\_error 队列中发送消息，该队列的消息由 ticket 服务负责消费。

@Transactional
    @RabbitListener(queues = "order:ticket_error")
    public void handleError(OrderDTO msg) {
        LOG.info("Get order error for ticket unlock:{}", msg);
        int count = ticketRepository.unMoveTicket(msg.getCustomerId(), msg.getTicketNum());
        if (count == 0) {
            LOG.info("Ticket already unlocked:", msg);
        }
        count = ticketRepository.unLockTicket(msg.getCustomerId(), msg.getTicketNum());
        if (count == 0) {
            LOG.info("Ticket already unmoved, or not moved:", msg);
        }
        rabbitTemplate.convertAndSend("order:fail", msg);
    }

当扣款失败的时候，做三件事：

1.  撤销票的转移，也就是把票的 owner 字段重新置为 null。
2.  撤销锁票，也就是把票的 lock\_user 字段重新置为 null。
3.  向 order:fail 队列发送订单失败的消息。

## 3.4.8 下单失败（order） ##

下单失败的处理在 order 服务中，有三种情况会向 order:fail 队列发送消息：

1.  锁票失败
2.  扣款失败（客户账户余额不足）
3.  订单超时

@Transactional
    @RabbitListener(queues = "order:fail")
    public void handleFailed(OrderDTO msg) {
        LOG.info("Get failed order:{}", msg);
        Order order;
        if (msg.getId() == null) {
            order = newOrder(msg);
            order.setReason("TICKET_LOCK_FAIL");
        } else {
            order = orderRepository.getById(msg.getId());
            if (msg.getStatus().equals("NOT_ENOUGH_DEPOSIT")) {
                order.setReason("NOT_ENOUGH_DEPOSIT");
            }
        }
        order.setStatus("FAIL");
        orderRepository.save(order);
    }

该方法的具体处理逻辑如下：

1.  首先查看是否有订单 id，如果连订单 id 都没有，就说明是锁票失败，给订单设置 reason 属性的值为TICKET\_LOCK\_FAIL。
2.  如果有订单 id，则根据 id 查询订单信息，并判断订单状态是否为 NOT\_ENOUGH\_DEPOSIT，这个表示扣款失败，如果订单状态是 NOT\_ENOUGH\_DEPOSIT，则设置失败的 reason 也为此。
3.  最后设置订单状态为 FAIL，然后更新数据库中的订单信息即可。

## 3.4.9 订单超时（order） ##

order 服务中还有一个定时任务，定时去数据库中捞取那些处理失败的订单，如下：

@Scheduled(fixedDelay = 10000L)
    public void checkInvalidOrder() {
        ZonedDateTime checkTime = ZonedDateTime.now().minusMinutes(1L);
        List<Order> orders = orderRepository.findAllByStatusAndCreatedDateBefore("NEW", checkTime);
        orders.stream().forEach(order -> {
            LOG.error("Order timeout:{}", order);
            OrderDTO dto = new OrderDTO();
            dto.setId(order.getId());
            dto.setTicketNum(order.getTicketNum());
            dto.setUuid(order.getUuid());
            dto.setAmount(order.getAmount());
            dto.setTitle(order.getTitle());
            dto.setCustomerId(order.getCustomerId());
            dto.setStatus("TIMEOUT");
            rabbitTemplate.convertAndSend("order:ticket_error", dto);
        });
    }

可以看到，这里是去数据库中捞取那些状态为 NEW 并且是 1 分钟之前的订单，根据前面的分析，当锁票成功后，就会将订单的状态设置为 NEW 并且存入数据库中。换言之，当锁票成功一分钟之后，这张票还没有卖掉，就设置订单超时，同时向 order:ticket\_error 队列发送一条消息，这条消息在 ticket 服务中被消费，最终完成撤销交票、撤销锁票等操作。

这就是大致的代码处理流程。

再来回顾一下前面那张图：

![c179aff533685a805ea3ce88ffcd4697.png][]

结合着代码来看这张图是不是就很容易懂了。

## 3.5 测试 ##

接下来我们来进行一个简单的测试。

先来一个订票失败的测试，如下：

![203cf420e5261b7dfe6befa8bcdf4239.png][]

由于用户只有 1000 块钱，这张票要 10000，所以购票必然失败。请求执行成功后，我们查看 order 表，多了如下一条记录：

![88a6eece9e9c92865fc04ee2e6010e39.png][]

可以看到，订单失败的理由就是账户余额不足。此时查看 ticket 和 user 表，发现都完好如初（如果需要，则已经反向补偿了）。

接下来我们手动给 ticket 表中 lock\_user 字段设置一个值，如下：

![e4b27039a4a7e5249cd370cfe5070b5c.png][]

这个表示这张票已经被人锁定了。

然后我们发起一次购票请求（这次可以把金额设置到合理范围，其实不设置也行，反正这次失败还没走到付款这一步）：

![2caa573d2f435a95569199ce21043fd9.png][]

请求发送成功后，接下来我们去查看 order 表，多了如下一条记录：

![115f592c5b53e01dfe957b2c011f5e23.png][]

可以看到，这次下单失败的理由是锁票失败。此时查看 ticket 和 user 表，发现都完好如初（如果需要，则已经反向补偿了）。

最后再来一次成功测试，先把 ticket 表中的 lock\_user 字段置空，然后发送如下请求：

![b3584c345a6d5779cdf8fea8bf677f35.png][]

这次购票成功，查看 ticket 表，发票已经票有所属：

![fa000d3a723ce90c85f207d85e579183.png][]

查看订单表：

![46506dcedbb47a8e64aa92ee17bd17a2.png][]

可以多了一条成功的购票记录。

查看用户表：

![40785c1695e55f3cec3fb6fe95c38490.png][]

用户账户已扣款。

查看支付记录表：

![28e866518cb88c4eee70e10bcd95e683.png][]

可以看到已经有了支付记录。

## 4. 总结 ##

整体上来说，上面这个案例，技术上并没有什么难的，复杂之处在于设计。一开始要设计好消息的处理流程以及消息处理失败后如何进行补偿，这个是比较考验大家技术的。

另外上面案例中，消息的发送和消费都用到了 RabbitMQ 中的事务机制（确保消息消费成功）以及 Spring 中的事务机制（确保消息发送和数据保存同时成功），这些我就不再赘述了。

总之，通过消息中间件处理分布式事务，这种方式通过牺牲数据的强一致性换取性能的大幅提升，但是实现这种方式的成本和复杂度是比较高的，使用时还要看实际业务情况。

![ffc81b41cb144ef5ad3474f65e22fc2f.jpeg][]

[1b6198bee0d72ab91f89cc294fdbf6c0.png]: https://img-blog.csdnimg.cn/img_convert/1b6198bee0d72ab91f89cc294fdbf6c0.png
[facfd332bef94aaeb4bb09d180e2cb45.png]: https://img-blog.csdnimg.cn/img_convert/facfd332bef94aaeb4bb09d180e2cb45.png
[62760db00deb5beef9522aabf1c9eb56.png]: https://img-blog.csdnimg.cn/img_convert/62760db00deb5beef9522aabf1c9eb56.png
[c179aff533685a805ea3ce88ffcd4697.png]: https://img-blog.csdnimg.cn/img_convert/c179aff533685a805ea3ce88ffcd4697.png
[203cf420e5261b7dfe6befa8bcdf4239.png]: https://img-blog.csdnimg.cn/img_convert/203cf420e5261b7dfe6befa8bcdf4239.png
[88a6eece9e9c92865fc04ee2e6010e39.png]: https://img-blog.csdnimg.cn/img_convert/88a6eece9e9c92865fc04ee2e6010e39.png
[e4b27039a4a7e5249cd370cfe5070b5c.png]: https://img-blog.csdnimg.cn/img_convert/e4b27039a4a7e5249cd370cfe5070b5c.png
[2caa573d2f435a95569199ce21043fd9.png]: https://img-blog.csdnimg.cn/img_convert/2caa573d2f435a95569199ce21043fd9.png
[115f592c5b53e01dfe957b2c011f5e23.png]: https://img-blog.csdnimg.cn/img_convert/115f592c5b53e01dfe957b2c011f5e23.png
[b3584c345a6d5779cdf8fea8bf677f35.png]: https://img-blog.csdnimg.cn/img_convert/b3584c345a6d5779cdf8fea8bf677f35.png
[fa000d3a723ce90c85f207d85e579183.png]: https://img-blog.csdnimg.cn/img_convert/fa000d3a723ce90c85f207d85e579183.png
[46506dcedbb47a8e64aa92ee17bd17a2.png]: https://img-blog.csdnimg.cn/img_convert/46506dcedbb47a8e64aa92ee17bd17a2.png
[40785c1695e55f3cec3fb6fe95c38490.png]: https://img-blog.csdnimg.cn/img_convert/40785c1695e55f3cec3fb6fe95c38490.png
[28e866518cb88c4eee70e10bcd95e683.png]: https://img-blog.csdnimg.cn/img_convert/28e866518cb88c4eee70e10bcd95e683.png
[ffc81b41cb144ef5ad3474f65e22fc2f.jpeg]: https://img-blog.csdnimg.cn/ffc81b41cb144ef5ad3474f65e22fc2f.jpeg