哪種分布式事務處理方案效率最高？必然是...

作者：江南一點雨 2021-09-03 10:37:35

開發架構分布式

在微服務系統中，服務之間的互相調用，我們可以使用 HTTP 的方式，例如 OpenFeign，也可以使用 RPC 的方式，例如 Dubbo，除了這些方案之外，我們也可以使用消息驅動，這是一種典型的響應式系統設計方案。

[[421456]]

前面幾篇文章松哥和大家介紹了 Seata 中四種分布式事務處理方案，相信經過前面的幾篇文章的學習，大家對于 Seata 中的分布式事務已經非常了解了。還沒看過前面文章的小伙伴，可以先看一下：

五分鐘帶你體驗一把分布式事務!so easy!
看了那么多博客，還是不懂 TCC，不妨看看這個案例!
XA 事務水很深，小伙子我怕你把握不住!
你這 Saga 事務保“隔離性”嗎?

不過很多小伙伴看完后感覺 Seata 對于分布式事務的處理，代碼雖然簡單，但是內部花費在網絡上的時間消耗太多了，在高并發場景下，這似乎并不是一種很好的解決方案。

要說哪種分布式事務處理方案效率高，必然繞不開消息中間件!基于消息中間件的兩階段提交方案，通常用在高并發場景下。這種方式通過犧牲數據的強一致性換取性能的大幅提升，不過實現這種方式的成本和復雜度是比較高的，使用時還要看實際業務情況。

今天松哥想通過一個簡單的案例，來和大家聊一聊如何通過消息中間件來處理分布式事務。

1. 思路分析

先來說說整體思路。

有一個名詞叫做消息驅動的微服務，相信很多小伙伴都聽說過。怎么理解呢?

在消息驅動的微服務中，服務之間不再互相直接調用，當服務之間需要通信時，就把通信內容發送到消息中間件上，另一個服務則通過監聽消息中間件中的消息隊列，來完成相應的業務邏輯調用，過程就是這么個過程，并不難，具體怎么玩，我們繼續往下看。

2. 業務分析

折騰了半天，后來松哥在網上找到了一個別人寫好的例子，我覺得用來演示這個問題特別合適，所以我就沒有自己寫案例了，直接用別人的代碼，我們來逐個分析，跟前面講分布式事務 Seata 的方式一致。

首先我們來看如下一張流程圖，這是一個用戶購票的案例：

當用戶想要購買一張票時：

向新訂單隊列中寫入一條數據。
Order Service 負責消費這個隊列中的消息，完成訂單的創建，然后再向新訂單繳費隊列中寫入一條消息。
User Service 負責消費新訂單繳費隊列中的消息，在 User Service 中完成對用戶賬戶余額的劃扣，然后向新訂單轉移票隊列中寫入一條消息。
Ticket Service 負責消費新訂單轉移票隊列，在 Ticket Service 中完成票的轉移，然后發送一條消息給訂單完成隊列。
最后 Order Service 中負責監聽訂單完成隊列，處理完成后的訂單。

這就是一個典型的消息驅動微服務，也是一個典型的響應式系統。在這個系統中，一共有三個服務，分別是：

Order Service
User Service
Ticket Service

這三個服務之間不會進行任何形式的直接調用，大家有事都是直接發送到消息中間件，其他服務則從消息中間件中獲取自己想要的消息然后進行處理。

具體到我們的實踐中，則多了一個檢查票是否夠用的流程，如下圖：

創建訂單時，先由 Ticket 服務檢查票是否夠用，沒問題的話再繼續發起訂單的創建。其他過程我就不說了。

另外還需要注意，在售票系統中，由于每張票都不同，例如每張票可能有座位啥的，因此一張票在數據庫中往往是被設計成一條記錄。

3. 實踐

流程我已經說明白了，接下來我們就來看看具體的代碼實踐。

3.1 準備數據庫

首先我們準備三個數據庫，分別是：

javaboy_order：訂單庫，用戶創建訂單等操作，在這個數據庫中完成。
javaboy_ticket：票務庫，這個庫中保存著所有的票據信息，每一張票都是一條記錄，都保存在這個庫中。
javaboy_user：用戶庫，這里保存著用戶的賬戶余額以及付款記錄等信息。

每個庫中都有各自對應的表，為了操作方便，這些表不用自己創建，將來等項目啟動了，利用 JPA 自動創建即可。

3.2 項目概覽

我們先來整體上看下這個項目，公眾號后臺回復 mq_tran 可以下載完整代碼：

一共有五個服務：

eureka：注冊中心
order：訂單服務
service：公共模塊
ticket：票務服務
user：用戶服務

下面分別來說。

3.3 注冊中心

有人說，都消息驅動了，還要注冊中心干嘛?

消息驅動沒錯，消息驅動微服務之后每個服務只管把消息往消息中間件上扔，每個服務又只管消費消息中間件上的消息，這個時候對于服務注冊中心似乎不是那么強需要。不過在我們這個案例中，消息驅動主要用來處理事務問題，其他常規需求我們還是用 OpenFeign 來處理，所以這里我們依然需要一個注冊中心。

這里的注冊中心我就選擇常見的 Eureka，省事一些。由于本文主要是和大家聊分布式事務，所以涉及到微服務的東西我就簡單介紹下，不會占用過多篇幅，如果大家還不熟悉 Spring Cloud 的用法，可以在公眾號后臺回復 vhr 有一套視頻介紹。

服務注冊中心的創建記得加上 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); 
}

這里的執行步驟如下：

首先根據訂單 id 去查找付款信息，檢查當前訂單是否已經完成付款，如果已經完成服務，則直接 return，這一步也是為了處理冪等性問題。
根據顧客的 id，查找到顧客的完整信息，包括顧客的賬戶余額。
檢查顧客的賬戶余額是否足夠支付票價，如果不夠，則設置 msg 的 status 為 NOT_ENOUGH_DEPOSIT，同時向 order:ticket_error 隊列發送消息，表示訂票失敗。
如果顧客賬戶余額足夠支付票價，則創建一個 PayInfo 對象，設置相關的支付信息，并存入 pay_info 表中。
調用 charge 方法完成顧客賬戶余額的扣款。
發送消息到 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); 
}

當扣款失敗的時候，做三件事：

撤銷票的轉移，也就是把票的 owner 字段重新置為 null。
撤銷鎖票，也就是把票的 lock_user 字段重新置為 null。
向 order:fail 隊列發送訂單失敗的消息。

3.4.8 下單失敗(order)

下單失敗的處理在 order 服務中，有三種情況會向 order:fail 隊列發送消息：

鎖票失敗
扣款失敗(客戶賬戶余額不足)
訂單超時

@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); 
}

該方法的具體處理邏輯如下：

首先查看是否有訂單 id，如果連訂單 id 都沒有，就說明是鎖票失敗，給訂單設置 reason 屬性的值為TICKET_LOCK_FAIL。
如果有訂單 id，則根據 id 查詢訂單信息，并判斷訂單狀態是否為 NOT_ENOUGH_DEPOSIT，這個表示扣款失敗，如果訂單狀態是 NOT_ENOUGH_DEPOSIT，則設置失敗的 reason 也為此。
最后設置訂單狀態為 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 服務中被消費，最終完成撤銷交票、撤銷鎖票等操作。

這就是大致的代碼處理流程。

再來回顧一下前面那張圖：