好看的小说完本推荐,怎样写网络小说

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案

如何用RabbitMQ解決分布式事務(wù)？

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

在棲霞等地區(qū)，都構(gòu)建了全面的區(qū)域性戰(zhàn)略布局，加強發(fā)展的系統(tǒng)性、市場前瞻性、產(chǎn)品創(chuàng)新能力，以專注、極致的服務(wù)理念，為客戶提供網(wǎng)站制作、成都網(wǎng)站制作網(wǎng)站設(shè)計制作按需設(shè)計網(wǎng)站,公司網(wǎng)站建設(shè),企業(yè)網(wǎng)站建設(shè),成都品牌網(wǎng)站建設(shè),成都全網(wǎng)營銷,外貿(mào)營銷網(wǎng)站建設(shè),棲霞網(wǎng)站建設(shè)費用合理。

1. 思路分析

先來說說整體思路。

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

在微服務(wù)系統(tǒng)中，服務(wù)之間的互相調(diào)用，我們可以使用 HTTP 的方式，例如 OpenFeign，也可以使用 RPC 的方式，例如 Dubbo，除了這些方案之外，我們也可以使用消息驅(qū)動，這是一種典型的響應(yīng)式系統(tǒng)設(shè)計方案。

在消息驅(qū)動的微服務(wù)中，服務(wù)之間不再互相直接調(diào)用，當(dāng)服務(wù)之間需要通信時，就把通信內(nèi)容發(fā)送到消息中間件上，另一個服務(wù)則通過監(jiān)聽消息中間件中的消息隊列，來完成相應(yīng)的業(yè)務(wù)邏輯調(diào)用，過程就是這么個過程，并不難，具體怎么玩，我們繼續(xù)往下看。

2. 業(yè)務(wù)分析

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

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

當(dāng)用戶想要購買一張票時：

向新訂單隊列中寫入一條數(shù)據(jù)。
Order Service 負責(zé)消費這個隊列中的消息，完成訂單的創(chuàng)建，然后再向新訂單繳費隊列中寫入一條消息。
User Service 負責(zé)消費新訂單繳費隊列中的消息，在 User Service 中完成對用戶賬戶余額的劃扣，然后向新訂單轉(zhuǎn)移票隊列中寫入一條消息。
Ticket Service 負責(zé)消費新訂單轉(zhuǎn)移票隊列，在 Ticket Service 中完成票的轉(zhuǎn)移，然后發(fā)送一條消息給訂單完成隊列。
最后 Order Service 中負責(zé)監(jiān)聽訂單完成隊列，處理完成后的訂單。

這就是一個典型的消息驅(qū)動微服務(wù)，也是一個典型的響應(yīng)式系統(tǒng)。在這個系統(tǒng)中，一共有三個服務(wù)，分別是：

Order Service
User Service
Ticket Service

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

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

創(chuàng)建訂單時，先由 Ticket 服務(wù)檢查票是否夠用，沒問題的話再繼續(xù)發(fā)起訂單的創(chuàng)建。其他過程我就不說了。

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

3. 實踐

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

3.1 準(zhǔn)備數(shù)據(jù)庫

首先我們準(zhǔn)備三個數(shù)據(jù)庫，分別是：

javaboy_order：訂單庫，用戶創(chuàng)建訂單等操作，在這個數(shù)據(jù)庫中完成。

javaboy_ticket：票務(wù)庫，這個庫中保存著所有的票據(jù)信息，每一張票都是一條記錄，都保存在這個庫中。

javaboy_user：用戶庫，這里保存著用戶的賬戶余額以及付款記錄等信息。

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

3.2 項目概覽

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

一共有五個服務(wù)：

eureka：注冊中心
order：訂單服務(wù)
service：公共模塊
ticket：票務(wù)服務(wù)
user：用戶服務(wù)
下面分別來說。

3.3 注冊中心

有人說，都消息驅(qū)動了，還要注冊中心干嘛？

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

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

服務(wù)注冊中心的創(chuàng)建記得加上 Spring Security，將自己的服務(wù)注冊中心保護起來。

這塊有一個小小的細節(jié)和大家多說兩句。

Eureka 用 Spring Security 保護起來之后，以后其他服務(wù)注冊都是通過 Http Basic 來認證，所以我們要在代碼中開啟 Http Basic 認證，如下（以前舊版本不需要下面這段代碼，但是新版本需要）：

3.4 購票服務(wù)

接下來我們就來看看購票服務(wù)。

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

購票是從下訂單開始，所以我們就先從訂單服務(wù) order 開始整個流程的分析。

3.4.1 新訂單處理（order）

當(dāng)用戶發(fā)起一個購票請求后，這個請求發(fā)送到 order 服務(wù)上，order 服務(wù)首先會向 order:new 隊列發(fā)送一條消息，開啟一個訂單的處理流程。代碼如下：

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

上面設(shè)置的 UUID 是整個訂單在處理過程中的一個唯一標(biāo)志符，也算是一條主線。

order:new? 隊列中的消息將被 ticket 服務(wù)消費，ticket 服務(wù)消費 order:new? 中的消息，并進行鎖票操作（鎖票的目的防止有兩個消費同時購買同一張票），鎖票成功后，ticket 服務(wù)將向 order:locked? 隊列發(fā)送一條消息，表示鎖票成功；否則向 order:fail 隊列發(fā)送一條消息表示鎖票失敗。

這里的 OrderDTO 對象將貫穿整個購票過程。

3.4.2 鎖票（ticket）

鎖票操作是在 ticket 服務(wù)中完成的，代碼如下：

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

先調(diào)用 lockTicket 方法去數(shù)據(jù)庫中鎖票，所謂的鎖票就是將要購買的票的 lock_user 字段設(shè)置為 customer_id（購買者的 id）。

如果鎖票成功（即數(shù)據(jù)庫修改成功），設(shè)置 msg 的狀態(tài)為 TICKET_LOCKED?，同時發(fā)送消息到 order:locked 隊列，表示鎖票成功。

如果鎖票失?。磾?shù)據(jù)庫修改失敗），設(shè)置 msg 的狀態(tài)為 TICKET_LOCK_FAIL?，同時發(fā)送消息到 order:fail 隊列，表示鎖票失敗。

3.4.2 鎖票成功（order）

接下來，由 order 服務(wù)消費 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);
}

鎖票成功后，先根據(jù)訂單的 UUID 去訂單數(shù)據(jù)庫查詢，是否已經(jīng)有訂單記錄了，如果有，說明這條消息已經(jīng)被處理了，可以防止訂單的重復(fù)處理（這塊主要是解決冪等性問題）。

如果訂單還沒有被處理，則創(chuàng)建一個新的訂單對象，并保存到數(shù)據(jù)庫中，創(chuàng)建新訂單對象的時候，需要設(shè)置訂單的 status 為 NEW。

最后設(shè)置 msg 的 status 為 NEW，然后向 order:pay? 隊列發(fā)送一條消息開啟付款流程，付款是由 user 服務(wù)提供的。user 服務(wù)中會檢查用戶的賬戶余額是否夠用，如果不夠用，就會發(fā)送消息到 order:ticket_error? 隊列，表示訂票失?。蝗绻囝~夠用，則進行正常的付款操作，并在付款成功后發(fā)送消息到 order:ticket_move 隊列，開啟票的轉(zhuǎn)移。

3.4.3 繳費（user）

鎖票成功后，接下來就是付費了，付費服務(wù)由 user 提供。

@Transactional
@RabbitListener(queues = "order:pay")
public void handle(OrderDTO msg) {
    LOG.info("Get new order to pay:{}", msg);
    // 先檢查payInfo判斷重復(fù)消息。
    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);
}

這里的執(zhí)行步驟如下：

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

調(diào)用 moveTicket 方法完成交票操作，也就是設(shè)置 ticket 表中票的 owner 為 customerId。

交票成功后，發(fā)送消息到 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);
}

這里的處理就比較簡單，訂單完成后，就設(shè)置訂單的狀態(tài)為 FINISH 即可。

上面介紹的是一條主線，順利的話，消息順著這條線走一遍，一個訂單就處理完成了。

不順利的話，就有各種幺蛾子，我們分別來看。

3.4.6 鎖票失?。╫rder）

鎖票是在 ticket 服務(wù)中完成的，如果鎖票失敗，就會直接向 order:fail 隊列發(fā)送消息，該隊列的消息由 order 服務(wù)負責(zé)消費。

3.4.7 扣款失?。╰icket）

扣款操作是在 user 中完成的，扣款失敗就會向 order:ticket_error 隊列中發(fā)送消息，該隊列的消息由 ticket 服務(wù)負責(zé)消費。

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

當(dāng)扣款失敗的時候，做三件事：

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

3.4.8 下單失敗（order）

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

鎖票失敗
扣款失?。蛻糍~戶余額不足）
訂單超時

@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 都沒有，就說明是鎖票失敗，給訂單設(shè)置 reason 屬性的值為TICKET_LOCK_FAIL。
如果有訂單 id，則根據(jù) id 查詢訂單信息，并判斷訂單狀態(tài)是否為NOT_ENOUGH_DEPOSIT?，這個表示扣款失敗，如果訂單狀態(tài)是NOT_ENOUGH_DEPOSIT，則設(shè)置失敗的 reason 也為此。
最后設(shè)置訂單狀態(tài)為 FAIL，然后更新數(shù)據(jù)庫中的訂單信息即可。

3.4.9 訂單超時（order）

order 服務(wù)中還有一個定時任務(wù)，定時去數(shù)據(jù)庫中撈取那些處理失敗的訂單，如下：

@Scheduled(fixedDelay = 10000L)
public void checkInvalidOrder() {
    ZonedDateTime checkTime = ZonedDateTime.now().minusMinutes(1L);
    List 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);
    });
}

可以看到，這里是去數(shù)據(jù)庫中撈取那些狀態(tài)為 NEW 并且是 1 分鐘之前的訂單，根據(jù)前面的分析，當(dāng)鎖票成功后，就會將訂單的狀態(tài)設(shè)置為 NEW 并且存入數(shù)據(jù)庫中。換言之，當(dāng)鎖票成功一分鐘之后，這張票還沒有賣掉，就設(shè)置訂單超時，同時向 order:ticket_error 隊列發(fā)送一條消息，這條消息在 ticket 服務(wù)中被消費，最終完成撤銷交票、撤銷鎖票等操作。

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

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

結(jié)合著代碼來看這張圖是不是就很容易懂了。

3.5 測試

接下來我們來進行一個簡單的測試。

先來一個訂票失敗的測試，如下：

由于用戶只有 1000 塊錢，這張票要 10000，所以購票必然失敗。請求執(zhí)行成功后，我們查看 order 表，多了如下一條記錄：

可以看到，訂單失敗的理由就是賬戶余額不足。此時查看 ticket 和 user 表，發(fā)現(xiàn)都完好如初（如果需要，則已經(jīng)反向補償了）。

接下來我們手動給 ticket 表中 lock_user 字段設(shè)置一個值，如下：

這個表示這張票已經(jīng)被人鎖定了。

然后我們發(fā)起一次購票請求（這次可以把金額設(shè)置到合理范圍，其實不設(shè)置也行，反正這次失敗還沒走到付款這一步）：

請求發(fā)送成功后，接下來我們?nèi)ゲ榭?order 表，多了如下一條記錄：

可以看到，這次下單失敗的理由是鎖票失敗。此時查看 ticket 和 user 表，發(fā)現(xiàn)都完好如初（如果需要，則已經(jīng)反向補償了）。

最后再來一次成功測試，先把 ticket 表中的 lock_user 字段置空，然后發(fā)送如下請求：

這次購票成功，查看 ticket 表，發(fā)票已經(jīng)票有所屬：

查看訂單表：

可以多了一條成功的購票記錄。

查看用戶表：

用戶賬戶已扣款。

查看支付記錄表：

可以看到已經(jīng)有了支付記錄。

4. 總結(jié)

整體上來說，上面這個案例，技術(shù)上并沒有什么難的，復(fù)雜之處在于設(shè)計。一開始要設(shè)計好消息的處理流程以及消息處理失敗后如何進行補償，這個是比較考驗大家技術(shù)的。

另外上面案例中，消息的發(fā)送和消費都用到了 RabbitMQ 中的事務(wù)機制（確保消息消費成功）以及 Spring 中的事務(wù)機制（確保消息發(fā)送和數(shù)據(jù)保存同時成功），這些我就不再贅述了。

總之，通過消息中間件處理分布式事務(wù)，這種方式通過犧牲數(shù)據(jù)的強一致性換取性能的大幅提升，但是實現(xiàn)這種方式的成本和復(fù)雜度是比較高的，使用時還要看實際業(yè)務(wù)情況。

網(wǎng)頁名稱：如何用RabbitMQ解決分布式事務(wù)？
標(biāo)題URL：http://fisionsoft.com.cn/article/cdgpsph.html