@Transactional用法詳解與事務避坑指南

更新時間：2025年10月23日 10:49:56 作者：-朽木

這篇文章主要介紹了@Transactional用法與事務避坑指南,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

需求：業(yè)務單位需要觀測訂單動態(tài)，所以要在訂單的各個狀態(tài)節(jié)點上傳狀態(tài)至狀態(tài)池系統(tǒng)
方案：又是Easy的需求，不就是在每個訂單狀態(tài)改變時觸發(fā)上傳狀態(tài)，為了不影響訂單正常流轉，增加一個消息中間件MQ

前言

在Java中，@Transactional是一個非常重要的注解，用于聲明事務管理的行為。

它可以被應用在類級別或方法級別上，并且提供了多種選項來控制事務的傳播行為、隔離級別、超時設置和回滾條件等。

一、Java的@Transactional解釋

@Transactional 注解在 Java 中用于聲明式事務管理，它可以應用于接口、接口方法、類以及類的方法上。在默認配置下，Spring 中的 @Transactional 注解會使得方法在事務的上下文中執(zhí)行，其事務邊界默認為方法開始執(zhí)行時開始，方法正常結束時提交事務，方法執(zhí)行過程中拋出異常時回滾事務。

但是，你可以通過在 @Transactional 注解中設置 propagation 屬性來改變事務的傳播行為，例如可以設置為Propagation.MANDATORY，這意味著該方法必須在一個已經(jīng)存在的事務中執(zhí)行，否則就會拋出異常。

另外，你可以通過設置 phase 屬性來改變事務的提交時機，例如可以設置為Phase.AFTER_COMPLETION，這樣事務就會在整個事務完成后提交，不論事務是正常結束還是異常結束。

以下是一個使用 @Transactional 注解的示例，其中 propagation 設置為Propagation.REQUIRED（默認值），表示如果當前存在事務，則加入該事務；如果不存在，則創(chuàng)建一個新的事務。

import org.springframework.transaction.annotation.Transactional;
import org.springframework.stereotype.Service;
 
@Service
public class MyService {
 
    @Transactional
    public void someTransactionalMethod() {
        // 方法執(zhí)行的代碼
    }
}

二、踩坑

接到的需求中：每個訂單狀態(tài)改變時觸發(fā)上傳狀態(tài)，為了不影響訂單正常流轉，增加一個消息中間件MQ

流程圖如下：

二、狀態(tài)統(tǒng)一入口：推送

    private SendResult sendStatusToMq(SendStatusPoolMsgVo sendStatusPoolMsgVo) {
        String msg = JSON.toJSONString(sendStatusPoolMsgVo, SerializerFeature.PrettyFormat,
                SerializerFeature.WriteMapNullValue);
        LOGGER.info("sendStatusPoolMsgToMq,發(fā)送的消息{}", msg);
        Message<String> message = MessageBuilder.withPayload(msg).build();
        SendResult sendResult = new SendResult();
        try {
            sendResult = rocketMQTemplate.syncSend(statusPoolTopic, message, 3000, 2);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("sendStatusToMq err", e);
            TaskMqModel taskMqModel = new TaskMqModel();
            taskMqModel.setTopic(statusPoolTopic);
            taskMqModel.setMessageType(MessageTypeConstant.MQ_SUPPLE_STATUSPOOL);
            taskMqModel.setDisposeFlag(RequestConstant.REQUEST_ORIGINAL);
            taskMqModel.setRawMessage(msg);
            taskMqModel.setCreateTime(new Date());
            taskMqService.save(taskMqModel);
        }
        return sendResult;
    }

我這里做了個狀態(tài)統(tǒng)一入口，同時為了避免推送MQ服務失敗，做了個異常捕獲保存推送消息，方便重推。

三、MQ消費

// ... 其他業(yè)務邏輯 ...
        // 獲取訂單信息
        String orderNo = msgVo.getOrderNo();
        orderModel orderModel = orderService.getOrderByOrderNo(orderNo);
        //如果是客戶下單再去待處理訂單查詢
        if (Objects.isNull(fmOrderModel)) {
            LOGGER.err("未查到信息，初始化失??！");
            return;
        }
//

然后我的日志全是：

[2024-08-8 08:36:50.161][ERROR][ConsumeMessageThread_2] 未查到信息，初始化失?。?
[2024-08-8 08:37:57.015][ERROR][ConsumeMessageThread_2] 未查到信息，初始化失敗！
[2024-08-8 08:42:45.536][ERROR][ConsumeMessageThread_2] 未查到信息，初始化失??！

我查看數(shù)據(jù)庫是完全能查到的，有點奇怪????????

四、問題與解決

在Java中，@Transactional 注解通常用于聲明方法執(zhí)行的事務邊界，確保方法內的一系列操作要么全部成功，要么全部失敗。它通常與Spring框架的事務管理器一起使用，以提供聲明式的事務管理。

如果你的訂單數(shù)據(jù)保存操作使用了 @Transactional，并且在保存數(shù)據(jù)之后立即推送消息到消息隊列（MQ），那么消費者可能查不到數(shù)據(jù)的原因可能與事務的隔離級別有關。

下面是我搜到的一些可能導致該問題的原因：

事務隔離級別：如果你的數(shù)據(jù)庫事務使用了較高的隔離級別（如可重復讀或串行化），那么在事務提交之前，其他事務（包括MQ消費者）可能看不到該事務所做的更改。
事務傳播行為：如果MQ消費者也運行在Spring管理的事務中，并且事務的傳播行為設置為不支持當前事務（例如，Propagation.NOT_SUPPORTED），那么消費者可能會在不同的事務上下文中運行，從而看不到未提交的更改。
延遲提交：如果事務在消息發(fā)送之后才提交，MQ消費者可能會在事務提交之前讀取數(shù)據(jù)庫，因此看不到新數(shù)據(jù)。
MQ消息延遲：如果消息隊列本身存在延遲，消費者可能會在數(shù)據(jù)還未寫入數(shù)據(jù)庫時就接收到消息。
數(shù)據(jù)庫緩存：某些數(shù)據(jù)庫實現(xiàn)可能會緩存數(shù)據(jù)，這可能導致即使數(shù)據(jù)已經(jīng)寫入，消費者也無法立即看到更新。
MQ消費者處理邏輯：如果MQ消費者在處理消息時沒有正確地處理事務或數(shù)據(jù)庫查詢，也可能導致查不到數(shù)據(jù)。

解決方式

在Spring框架中，我通過實現(xiàn)TransactionSynchronization接口來在事務提交后執(zhí)行回調操作。

以下是實現(xiàn)事務提交回調發(fā)送消息的方法：

@Component
public class TransactionalMessageSender implements TransactionSynchronization {

    @Autowired
    private SomeMessageService messageService; // 消息發(fā)送服務

    @Override
    public void beforeCommit(boolean readOnly) {
        // 這里可以執(zhí)行一些操作，但事務還未提交
    }

    @Override
    public void beforeCompletion() {
        // 事務即將提交，可以在這里準備發(fā)送消息
    }

    @Override
    public void afterCommit() {
        // 事務已經(jīng)提交，可以在這里發(fā)送消息
        sendResult = rocketMQTemplate.syncSend(statusPoolTopic, message, 3000, 2);
    }

    @Override
    public void afterCompletion(TransactionStatus status) {
        // 事務已經(jīng)完成，無論是提交還是回滾
        if (status == TransactionStatus.COMMITTED) {
            // 這里可以執(zhí)行一些事務提交后的清理工作
        }
    }
}