Seata之分布式事务问题及解决方案

<!-- Spring Cloud Alibaba 依赖 -->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
</dependency>

-- 每个业务数据库均需执行
CREATE TABLE IF NOT EXISTS `undo_log`
(
    `id`            BIGINT(20)   NOT NULL AUTO_INCREMENT,
    `branch_id`     BIGINT(20)   NOT NULL,
    `xid`           VARCHAR(100) NOT NULL,
    `context`       VARCHAR(128) NOT NULL,
    `rollback_info` LONGBLOB     NOT NULL,
    `log_status`    INT(11)      NOT NULL,
    `log_created`   DATETIME     NOT NULL,
    `log_modified`  DATETIME     NOT NULL,
    PRIMARY KEY (`id`),
    UNIQUE KEY `ux_undo_log` (`xid`, `branch_id`)
) ENGINE = InnoDB
  AUTO_INCREMENT = 1
  DEFAULT CHARSET = utf8;

<dependency>
  <groupId>com.alibaba.cloud</groupId>
  <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
</dependency>

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

transport {
  # tcp, Unix-domain-socket
  type = "TCP"
  #NIO, NATIVE
  server = "NIO"
  #enable heartbeat
  heartbeat = true
  # the tm client batch send request enable
  enableTmClientBatchSendRequest = false
  # the rm client batch send request enable
  enableRmClientBatchSendRequest = true
   # the rm client rpc request timeout
  rpcRmRequestTimeout = 2000
  # the tm client rpc request timeout
  rpcTmRequestTimeout = 30000
  # the rm client rpc request timeout
  rpcRmRequestTimeout = 15000
  #thread factory for netty
  threadFactory {
    bossThreadPrefix = "NettyBoss"
    workerThreadPrefix = "NettyServerNIOWorker"
    serverExecutorThread-prefix = "NettyServerBizHandler"
    shareBossWorker = false
    clientSelectorThreadPrefix = "NettyClientSelector"
    clientSelectorThreadSize = 1
    clientWorkerThreadPrefix = "NettyClientWorkerThread"
    # netty boss thread size
    bossThreadSize = 1
    #auto default pin or 8
    workerThreadSize = "default"
  }
  shutdown {
    # when destroy server, wait seconds
    wait = 3
  }
  serialization = "seata"
  compressor = "none"
}
service {
  #transaction service group mapping
  vgroupMapping.default_tx_group = "default"
  #only support when registry.type=file, please don't set multiple addresses
  default.grouplist = "127.0.0.1:8091"
  #degrade, current not support
  enableDegrade = false
  #disable seata
  disableGlobalTransaction = false
}

client {
  rm {
    asyncCommitBufferLimit = 10000
    lock {
      retryInterval = 10
      retryTimes = 30
      retryPolicyBranchRollbackOnConflict = true
    }
    reportRetryCount = 5
    tableMetaCheckEnable = false
    tableMetaCheckerInterval = 60000
    reportSuccessEnable = false
    sagaBranchRegisterEnable = false
    sagajsonParser = "fastjson"
    sagaRetryPersistModeUpdate = false
    phpsagaCompensatePersistModeUpdate = false
    tccActionInterceptorOrder = -2147482648 #Ordered.HIGHEST_PRECEDENCE + 1000
    sqlParserType = "druid"
    branchExecutionTimeoutXA = 60000
    connectionTwoPhaseHoldTimeoutXA = 10000
  }
  tm {
    commitRetryCount = 5
    rollbackRetryCount = 5
    defaultGlobalTransactionTimeout = 60000
    degradeCheck = false
    degradeCheckPeriod = 2000
    degradeCheckAllowTimes = 10
    interceptorOrder = -2147482648 #Ordered.HIGHEST_PRECEDENCE + 1000
  }
  undo {
    dataValidation = true
    onlyCareUpdateColumns = true
    logSerialization = "jackson"
    logTable = "undo_log"
    compress {
      enable = true
      # allow zip, gzip, deflater, lz4, bzip2, zstd default is zip
      type = zip
      # if rollback info size > threshold, then will be compress
      # allow k m g t
      threshold = 64k
    }
  }
  loadBalance {
      type = "XID"
      virtualNodes = 10
  }
}
log {
  exceptionRate = 100
}
tcc {
  fence {
    # tcc fence log table name
    logTableName = tcc_fence_log
    # tcc fence log clean period
    cleanPeriod = 1h
  }
}

@Service
public class OrderService {
    @GlobalTransactional(name = "createOrder", timeoutMills = 60000)
    public void createOrder(OrderDTO order) {
        // 1. 扣减库存（调用库存服务）
        storageFeignClient.deduct(order.getProductId());
        // 2. 创建订单（本地事务）
        orderMapper.insert(order);
        // 3. 模拟异常触发回滚
        int i = 1 / 0; 
    }
}

seata:
  data-source-proxy-mode: XA  # 默认AT

// 1. 定义TCC接口
public interface StorageTccService {
    @TwoPhaseBusinessAction(name = "deduct", 
                          commitMethod = "confirm", 
                          rollbackMethod = "cancel")
    boolean deduct(@BusinessActionContextParameter(paramName = "productId") String productId,
                   @BusinessActionContextParameter(paramName = "count") Integer count);
 android   
    boolean confirm(BusinessActionContext context);
    boolean cancel(BusinessActionContext context);
}

// 2. 实现Try逻辑
@Service
public class StorageTccServiceImpl implements StorageTccService {
    @Override
    public boolean deduct(String productId, Integer count) {
        // Try阶段：资源预留（例如冻结库存）
        return storageMapper.freezeStock(productId, count) > 0;
    }
    
    @Override
    public boolean confirm(BusinessActionContext context) {
        // Confirm阶段：真实扣减（例如删除冻结记录）
        String productId = context.getActionContext("productId");
  android      Integer count = context.getActionContext("count");
        return storageMapper.reduceStock(productId, count) > 0;
    }
    
    @Override
    public boolean cancel(BusinessActionContext context) {
        // Cancel阶段：释放资源（例如恢复冻结库存）
        String productId = context.getActionContext("productId");
        Integer count = context.getActionContext("count");
        return storageMappandroider.unfreezeStock(productId, count) > 0;
    }
}

@SagaStart
public void createOrderSaga(Order order) {
    // 1. 创建订单
    orderService.create(order);
    // 2. 调用支付服务（若失败则触发逆向操作）
    paymentService.pay(order.getId());
}
// 定义补偿方法
@Compensate
public void compensateOrder(Order order) {
    orderService.delete(order.getId());
}