TX-LCN分布式事務之TCC模式

什么是TCC模式

TCC模式是TX-LCN分布式事務模式的一種，T-try-嘗試執行業務、C-confirm-確認執行業務、 C-cancel-取消執行業務

原理

TCC事務機制相對於傳統事務機制（X/Open XA Two-Phase-Commit），其特征在於它不依賴資源管理器(RM)對XA的支持， 而是通過對（由業務系統提供的）業務邏輯的調度來實現分布式事務。主要由三步操作，Try: 嘗試執行業務、 Confirm:確認執行業務、 Cancel: 取消執行業務。

模式特點

• 該模式對代碼的嵌入性高，要求每個業務需要寫三種步驟的操作。
• 該模式對有無本地事務控制都可以支持使用面廣。
• 數據一致性控制幾乎完全由開發者控制，對業務開發難度要求高。

源碼解讀

首先我們來看幾個關鍵的類TransactionAspect事務切面類、DTXLogicWeaver分布式事務調度器、DTXServiceExecutor分布式事務執行器

TransactionAspect 作用

• 源碼

@Aspect
@Component
public class TransactionAspect implements Ordered {
    private static final Logger log = LoggerFactory.getLogger(TransactionAspect.class);
    private final TxClientConfig txClientConfig;
    private final DTXLogicWeaver dtxLogicWeaver;

    public TransactionAspect(TxClientConfig txClientConfig, DTXLogicWeaver dtxLogicWeaver) {
        this.txClientConfig = txClientConfig;
        this.dtxLogicWeaver = dtxLogicWeaver;
    }

    @Pointcut("@annotation(com.codingapi.txlcn.tc.annotation.TccTransaction)")
    public void tccTransactionPointcut() {
    }
    
    @Around("tccTransactionPointcut() && !lcnTransactionPointcut()&& !txcTransactionPointcut() && !txTransactionPointcut()")
    public Object runWithTccTransaction(ProceedingJoinPoint point) throws Throwable {
        DTXInfo dtxInfo = DTXInfo.getFromCache(point);
        TccTransaction tccTransaction = (TccTransaction)dtxInfo.getBusinessMethod().getAnnotation(TccTransaction.class);
        dtxInfo.setTransactionType("tcc");
        dtxInfo.setTransactionPropagation(tccTransaction.propagation());
        DTXLogicWeaver var10000 = this.dtxLogicWeaver;
        point.getClass();
        return var10000.runTransaction(dtxInfo, point::proceed);
    }

    public int getOrder() {
        return this.txClientConfig.getDtxAspectOrder();
    }

由該類的源碼，我們能夠明白，通過解析@TccTransaction注解進行相應的操作。代碼會調用到DTXLogicWeaver類

DTXLogicWeaver 作用

    public Object runTransaction(DTXInfo dtxInfo, BusinessCallback business) throws Throwable {
        if (Objects.isNull(DTXLocalContext.cur())) {
            DTXLocalContext.getOrNew();
            log.debug("<---- TxLcn start ---->");
            DTXLocalContext dtxLocalContext = DTXLocalContext.getOrNew();
            TxContext txContext;
            if (this.globalContext.hasTxContext()) {
                txContext = this.globalContext.txContext();
                dtxLocalContext.setInGroup(true);
                log.debug("Unit[{}] used parent's TxContext[{}].", dtxInfo.getUnitId(), txContext.getGroupId());
            } else {
                txContext = this.globalContext.startTx();
            }

            if (Objects.nonNull(dtxLocalContext.getGroupId())) {
                dtxLocalContext.setDestroy(false);
            }

            dtxLocalContext.setUnitId(dtxInfo.getUnitId());
            dtxLocalContext.setGroupId(txContext.getGroupId());
            dtxLocalContext.setTransactionType(dtxInfo.getTransactionType());
            TxTransactionInfo info = new TxTransactionInfo();
            info.setBusinessCallback(business);
            info.setGroupId(txContext.getGroupId());
            info.setUnitId(dtxInfo.getUnitId());
            info.setPointMethod(dtxInfo.getBusinessMethod());
            info.setPropagation(dtxInfo.getTransactionPropagation());
            info.setTransactionInfo(dtxInfo.getTransactionInfo());
            info.setTransactionType(dtxInfo.getTransactionType());
            info.setTransactionStart(txContext.isDtxStart());
            boolean var15 = false;

            Object var6;
            try {
                var15 = true;
                var6 = this.transactionServiceExecutor.transactionRunning(info);
                var15 = false;
            } finally {
                if (var15) {
                    if (dtxLocalContext.isDestroy()) {
                        synchronized(txContext.getLock()) {
                            txContext.getLock().notifyAll();
                        }

                        if (!dtxLocalContext.isInGroup()) {
                            this.globalContext.destroyTx();
                        }

                        DTXLocalContext.makeNeverAppeared();
                        TracingContext.tracing().destroy();
                    }

                    log.debug("<---- TxLcn end ---->");
                }
            }

            if (dtxLocalContext.isDestroy()) {
                synchronized(txContext.getLock()) {
                    txContext.getLock().notifyAll();
                }

                if (!dtxLocalContext.isInGroup()) {
                    this.globalContext.destroyTx();
                }

                DTXLocalContext.makeNeverAppeared();
                TracingContext.tracing().destroy();
            }

            log.debug("<---- TxLcn end ---->");
            return var6;
        } else {
            return business.call();
        }
    }

以上代碼是該類的核心邏輯，可以看出來TX-LCN事務的處理全部都是走的這個類的該方法，最終會調用到DTXServiceExecutor分布式事務執行器

DTXServiceExecutor 作用

    /**
     * 事務業務執行
     *
     * @param info info
     * @return Object
     * @throws Throwable Throwable
     */
    public Object transactionRunning(TxTransactionInfo info) throws Throwable {

        // 1. 獲取事務類型
        String transactionType = info.getTransactionType();

        // 2. 獲取事務傳播狀態
        DTXPropagationState propagationState = propagationResolver.resolvePropagationState(info);

        // 2.1 如果不參與分布式事務立即終止
        if (propagationState.isIgnored()) {
            return info.getBusinessCallback().call();
        }

        // 3. 獲取本地分布式事務控制器
        DTXLocalControl dtxLocalControl = txLcnBeanHelper.loadDTXLocalControl(transactionType, propagationState);

        // 4. 織入事務操作
        try {
            // 4.1 記錄事務類型到事務上下文
            Set<String> transactionTypeSet = globalContext.txContext(info.getGroupId()).getTransactionTypes();
            transactionTypeSet.add(transactionType);

            dtxLocalControl.preBusinessCode(info);

            // 4.2 業務執行前
            txLogger.txTrace(
                    info.getGroupId(), info.getUnitId(), "pre business code, unit type: {}", transactionType);

            // 4.3 執行業務
            Object result = dtxLocalControl.doBusinessCode(info);

            // 4.4 業務執行成功
            txLogger.txTrace(info.getGroupId(), info.getUnitId(), "business success");
            dtxLocalControl.onBusinessCodeSuccess(info, result);
            return result;
        } catch (TransactionException e) {
            txLogger.error(info.getGroupId(), info.getUnitId(), "before business code error");
            throw e;
        } catch (Throwable e) {
            // 4.5 業務執行失敗
            txLogger.error(info.getGroupId(), info.getUnitId(), Transactions.TAG_TRANSACTION,
                    "business code error");
            dtxLocalControl.onBusinessCodeError(info, e);
            throw e;
        } finally {
            // 4.6 業務執行完畢
            dtxLocalControl.postBusinessCode(info);
        }
    }

通過以上代碼可以看出，該類是整個事務執行關鍵類。

以上就是TCC模式比較核心的代碼，其他的分支代碼就不一一贅述了。可以看到和LCN模式有一些代碼是重復的，整體就是通過解析不同的注解走不同的事務模式

實戰

由第一篇分布式事務之TX-LCN 我們規划了倆個TC分別是lcn-order 服務和lcn-pay服務，我們的思路是訂單服務調用支付服務，分別在訂單服務表t_order和支付服務表t_pay中插入插入數據。

訂單服務核心代碼和數據表腳本

• 代碼

/**
 * @author:triumphxx
 * @Date:2021/10/24
 * @Time:2:13 下午
 * @微信公眾號：北漂碼農有話說
 * @網站：http://blog.triumphxx.com.cn
 * @GitHub https://github.com/triumphxx
 * @Desc:
 **/
@RestController
public class TccOrderController {

    @Autowired
    TOrderDao tOrderDao;

    @Autowired
    private RestTemplate restTemplate;

    //保存主鍵信息
    private static Map<String,Integer> maps = new HashMap<>();


    @PostMapping("/add-order-tcc")
    @Transactional(rollbackFor = Exception.class)
    @TccTransaction
    public String add(){
        TOrder bean = new TOrder();
        Long no =  Math.round((Math.random()+1) * 1000);
        bean.setTId(no.intValue());
        bean.setTName("order"+no.intValue());
        JSONObject date = new JSONObject();
        date.put("tId",bean.getTId());
        date.put("tName",bean.getTName()+"pay");
        restTemplate.postForEntity("http://lcn-pay/add-pay-tcc",date,String.class);
//        int i = 1/0;
        tOrderDao.insert(bean);
        maps.put("a",no.intValue());
        System.out.println("本次新增訂單號為" + no.intValue());
        return "新增訂單成功";
    }

    public String confirmAdd(){
        System.out.println("order confirm ");
        System.out.println("訂單新增成功 id為:"+maps.get("a"));
        maps.clear();
        return "新增訂單成功";
    }
    /**
     * 逆sql
     * @param  
     * @return
     */
    public String cancelAdd(){
        Integer a = maps.get("a");
        System.out.println("order cancel ");
        System.out.println("刪除的訂單號為 :"+a);
        tOrderDao.deleteByPrimaryKey(a);
        maps.clear();
        return "新增訂單成功";
    }
}

代碼解讀：若使用TCC模式只需要在你的方法上添加注解@TccTransaction，相對於lcn模式，我們可以看到多出來很多代碼，所以tcc模式的數據一致性依賴於開發者來進行控制，對於開發要求高， 需要注意的是confirm的操作，方法名必須是confirm開頭，cancel操作同樣的道理，在confirm中我們可以做一些業務執行 成功后的操作，cancel操作的是try的逆操作，若try是新增數據，那么cancel就是刪除數據。注意代碼中有一個map的數據結果， 就是進行try、confirm、cancel操作業務的唯一標識進行相應的操作。

• 腳本

CREATE TABLE `t_order` (
   `t_id` int(11) NOT NULL,
   `t_name` varchar(45) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=latin1

支付服務核心代碼和數據表腳本

• 代碼

/**
 * @author:triumphxx
 * @Date:2021/10/24
 * @Time:2:26 下午
 * @微信公眾號：北漂碼農有話說
 * @網站：http://blog.triumphxx.com.cn
 * @GitHub https://github.com/triumphxx
 * @Desc:
 **/
@RestController
public class TccPayController {
    @Autowired
    TPayDao tPayDao;
    //保存主鍵信息
    private static Map<String,Integer> maps = new HashMap<>();

    @PostMapping("/add-pay-tcc")
    @Transactional(rollbackFor = Exception.class)
    @TccTransaction
    public String addPay(@RequestBody Map map){
        TPay tPay = new TPay();
        tPay.setTId((Integer) map.get("tId"));
        tPay.setTName((String) map.get("tName"));
        tPayDao.insert(tPay);
        maps.put("a",(Integer) map.get("tId"));
//        int i = 1/0;
        System.out.println("本次新增支付號為" + (Integer) map.get("tId"));
        return "新增支付成功";
    }

    public String confirmAddPay(Map map){
        System.out.println("pay confirm");
        System.out.println("本次新增支付號為" + maps.get("a"));
        maps.clear();
        return "新增支付成功";

    }


    /**
     * 逆sql
     * @param  map
     * @return
     */
    public String cancelAddPay(Map map){
        Integer a = maps.get("a");
        System.out.println("pay cancel");
        System.out.println("本次刪除支付號為" + maps.get("a"));
        tPayDao.deleteByPrimaryKey(a);
        maps.clear();
        return "取消支付成功";

    }
}

• 腳本

CREATE TABLE `t_pay` (
     `t_id` int(11) NOT NULL,
     `t_name` varchar(45) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=latin1

測試流程

• 啟動 Redis
• 啟動 TM
• 啟動注冊中心 eureka-server
• 啟動服務 lcn-order
• 啟動服務 lcn-pay
• 請求接口 http://localhost:8001/add-order
• 代碼搞一個異常看數據是否進行回滾

小結

本篇我們分析了TX-LCN分布式事務的TCC模式的原理及相關源碼，以及搭建服務的進行測試。希望能對大家有所幫助。 源碼地址源碼傳送門