幂等性控制

若第一阶段冻结数据失败,第二阶段的回滚操作也会执行,二阶段失败,TC会重复发送二阶段指令,模块会重复执行二阶段操作.将会导致数据库出现冻结的负数库存和增加不存在的库存数据(即冻结的负数库存)

幂等性控制就是让多次重复的操作和一次操作的结果相同.

  • 在第一阶段成果设置一个第一阶段成功标记,失败则没有标记

  • 在第二阶段执行前先进行检查标记是否存在,存在则可以执行第二阶段的提交或者回滚操作,二阶段执行成功后删除标记,没有标记则不执行第二阶段的操作.

    在库存服务和订单服务中添加幂等性控制

    创建幂等性工具类:

    1. public class ResultHolder {
    2.   private static Map<Class<?>, Map<String, String>> map = new ConcurrentHashMap<Class<?>, Map<String, String>>();
    4.   public static void setResult(Class<?> actionClass, String xid, String v) {
    5.       Map<String, String> results = map.get(actionClass);
    7.       if (results == null) {
    8.           synchronized (map) {
    9.               if (results == null) {
    10.                   results = new ConcurrentHashMap<>();
    11.                   map.put(actionClass, results);
    12.               }
    13.           }
    14.       }
    16.       results.put(xid, v);
    17.   }
    19.   public static String getResult(Class<?> actionClass, String xid) {
    20.       Map<String, String> results = map.get(actionClass);
    21.       if (results != null) {
    22.           return results.get(xid);
    23.       }
    25.       return null;
    26.   }
    28.   public static void removeResult(Class<?> actionClass, String xid) {
    29.       Map<String, String> results = map.get(actionClass);
    30.       if (results != null) {
    31.           results.remove(xid);
    32.       }
    33.   }
    34. }

    在TCC接口中添加幂等性控制

    1. @Component
    2. public class OrderTccActionImpl implements OrderTccAction {
    3.   private final OrderMapper orderMapper;
    5.   @Autowired
    6.   public OrderTccActionImpl(OrderMapper orderMapper) {
    7.       this.orderMapper = orderMapper;
    8.   }
    10.   @Transactional
    11.   @Override
    12.   public boolean prepare(BusinessActionContext businessActionContext,
    13.                          Long id, Long userId,
    14.                          Long productId, Integer count, BigDecimal money) {
    15.       orderMapper.createOrder(new Order(id, userId, productId, count, money, 0));
    16.       /*
    17.       幂等性控制
    18.       第一阶段成功时设置成功标记
    19.        */
    20.       ResultHolder.setResult(OrderTccAction.class, businessActionContext.getXid(), "p");
    21.       return true;
    22.   }
    24.   @Transactional
    25.   @Override
    26.   public synchronized boolean commit(BusinessActionContext businessActionContext) {
    27.       //先判断标记是否存在
    28.       if (ResultHolder.getResult(OrderTccAction.class, businessActionContext.getXid()) == null) {
    29.           return true;
    30.       }
    31.       Long orderId = Long.valueOf(businessActionContext.getActionContext("orderId").toString());
    32.       orderMapper.updateStatus(orderId, 1);
    33.       //二阶段成功则删除标记
    34.       ResultHolder.removeResult(OrderTccAction.class, businessActionContext.getXid());
    35.       return true;
    36.   }
    38.   @Transactional
    39.   @Override
    40.   public synchronized boolean rollback(BusinessActionContext businessActionContext) {
    41.       /*
    42.       一阶段失败则没有成功标记,回滚不执行
    43.       一阶段成功,有成功标记,但是其他模块执行失败,可以执行回归滚,回滚完成删除标记
    44.        */
    45.       if (ResultHolder.getResult(OrderTccAction.class, businessActionContext.getXid()) == null) {
    46.           return true;
    47.       }
    48.       Long orderId = Long.valueOf(businessActionContext.getActionContext("orderId").toString());
    49.       orderMapper.deleteById(orderId);
    50.       //二阶段成功则删除标记
    51.       ResultHolder.removeResult(OrderTccAction.class, businessActionContext.getXid());
    52.       return true;
    53.   }
    54. }

    账户服务中添加TCC事务

    添加TCC事务配置文件

  1. 添加registry.conf配置文件 ```powershell registry {

    file 、nacos 、eureka、redis、zk、consul、etcd3、sofa

    type = “eureka”

    nacos { serverAddr = “localhost” namespace = “” cluster = “default” } eureka {

    连接eureka，要从注册表发现 seata-server

    serviceUrl = “http://localhost:8761/eureka“

    application = “default”

    weight = “1”

    } redis { serverAddr = “localhost:6379” db = “0” password = “” cluster = “default” timeout = “0” } zk { cluster = “default” serverAddr = “127.0.0.1:2181” session.timeout = 6000 connect.timeout = 2000 username = “” password = “” } consul { cluster = “default” serverAddr = “127.0.0.1:8500” } etcd3 { cluster = “default” serverAddr = “http://localhost:2379“ } sofa { serverAddr = “127.0.0.1:9603” application = “default” region = “DEFAULT_ZONE” datacenter = “DefaultDataCenter” cluster = “default” group = “SEATA_GROUP” addressWaitTime = “3000” } file { name = “file.conf” } }

config {

file、nacos 、apollo、zk、consul、etcd3、springCloudConfig

type = “file”

nacos { serverAddr = “localhost” namespace = “” group = “SEATA_GROUP” } consul { serverAddr = “127.0.0.1:8500” } apollo { app.id = “seata-server” apollo.meta = “http://192.168.1.204:8801“ namespace = “application” } zk { serverAddr = “127.0.0.1:2181” session.timeout = 6000 connect.timeout = 2000 username = “” password = “” } etcd3 { serverAddr = “http://localhost:2379“ } file { name = “file.conf” } }

  2. 2. 添加file.conf配置文件
  3. ```powershell
  4. transport {
  5.   # tcp udt unix-domain-socket
  6.   type = "TCP"
  7.   #NIO NATIVE
  8.   server = "NIO"
  9.   #enable heartbeat
  10.   heartbeat = true
  11.   # the client batch send request enable
  12.   enableClientBatchSendRequest = true
  13.   #thread factory for netty
  14.   threadFactory {
  15.     bossThreadPrefix = "NettyBoss"
  16.     workerThreadPrefix = "NettyServerNIOWorker"
  17.     serverExecutorThread-prefix = "NettyServerBizHandler"
  18.     shareBossWorker = false
  19.     clientSelectorThreadPrefix = "NettyClientSelector"
  20.     clientSelectorThreadSize = 1
  21.     clientWorkerThreadPrefix = "NettyClientWorkerThread"
  22.     # netty boss thread size,will not be used for UDT
  23.     bossThreadSize = 1
  24.     #auto default pin or 8
  25.     workerThreadSize = "default"
  26.   }
  27.   shutdown {
  28.     # when destroy server, wait seconds
  29.     wait = 3
  30.   }
  31.   serialization = "seata"
  32.   compressor = "none"
  33. }
  34. service {
  35.   #transaction service group mapping
  36.   # order_tx_group 与 yml 中的 “tx-service-group: order_tx_group” 配置一致
  37.   # “seata-server” 与 TC 服务器的注册名一致
  38.   # 从eureka获取seata-server的地址，再向seata-server注册自己，设置group
  42.   # order_tx_group 事务组，对应使用哪个协调器
  43.   # seata-server 是注册表中的服务id
  44.   vgroupMapping.order_tx_group = "seata-server"
  49.   #only support when registry.type=file, please don't set multiple addresses
  50.   order_tx_group.grouplist = "127.0.0.1:8091"
  51.   #degrade, current not support
  52.   enableDegrade = false
  53.   #disable seata
  54.   disableGlobalTransaction = false
  55. }
  57. client {
  58.   rm {
  59.     asyncCommitBufferLimit = 10000
  60.     lock {
  61.       retryInterval = 10
  62.       retryTimes = 30
  63.       retryPolicyBranchRollbackOnConflict = true
  64.     }
  65.     reportRetryCount = 5
  66.     tableMetaCheckEnable = false
  67.     reportSuccessEnable = false
  68.   }
  69.   tm {
  70.     commitRetryCount = 5
  71.     rollbackRetryCount = 5
  72.   }
  73.   undo {
  74.     dataValidation = true
  75.     logSerialization = "jackson"
  76.     logTable = "undo_log"
  77.   }
  78.   log {
  79.     exceptionRate = 100
  80.   }
  81. }
  1. 修改yml配置添加事务组 ```powershell server: port: 8081

spring: application: name: account

datasource: url: jdbc:mysql:///seata_account?useUnicode=true&characterEncoding=UTF-8&serverTimezone=GMT%2B8 username: root password: root driver-class-name: com.mysql.cj.jdbc.Driver

添加事务组

cloud: alibaba: seata: tx-service-group: order_tx_group

eureka: client: service-url: defaultZone: http://localhost:8761/eureka

instance: prefer-ip-address: true instance-id: ${spring.cloud.client.ip-address}:${spring.application.name}:${server.port}

mybatis-plus: mapper-locations: classpath:mapper/*.xml type-aliases-package: cn.tedu.account.entity configuration: map-underscore-to-camel-case: true

logging: level: cn.tedu.account.mapper: debug

  1. <a name="PQZrD"></a>
  2. ### 修改Mapper添加事务处理
  3. ```java
  4. @Mapper
  5. public interface AccountMapper extends BaseMapper<Account> {
  6.     /**
  7.      * 扣除账户金额
  8.      *
  9.      * @param userId 用户id
  10.      * @param money 扣除金额数
  11.      */
  12.     void decrease(Long userId, BigDecimal money);
  14.     /**
  15.      * 查询账户
  16.      * @param userId 用户id
  17.      * @return 用户账户信息
  18.      */
  19.     Account findByUserId(Long userId);
  21.     /**
  22.      * 将可用金额冻结,对应第一阶段提交事务
  23.      * @param userId 用户id
  24.      * @param money 冻结金额数目
  25.      */
  26.     void updateResidueToFrozen(Long userId,BigDecimal money);
  28.     /**
  29.      * 冻结金额解冻变为已使用,对应第二阶段提交
  30.      * @param userId 用户id
  31.      * @param money 已使用金额
  32.      */
  33.     void updateFrozenToUsed(Long userId,BigDecimal money);
  35.     /**
  36.      * 冻结金额回滚至账户中,对应第二阶段回滚
  37.      * @param userId 用户id
  38.      * @param money 回滚金额
  39.      */
  40.     void updateFrozenToResidue(Long userId,BigDecimal money);
  41. }

xml文件如下:

  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <!DOCTYPE mapper
  3.         PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
  4.         "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
  5. <mapper namespace="cn.tedu.account.mapper.AccountMapper">
  6.     <resultMap id="BaseResultMap" type="Account">
  7.         <id column="id" property="id" jdbcType="BIGINT"/>
  8.         <result column="user_id" property="userId" jdbcType="BIGINT"/>
  9.         <result column="total" property="total" jdbcType="DECIMAL"/>
  10.         <result column="used" property="used" jdbcType="DECIMAL"/>
  11.         <result column="residue" property="residue" jdbcType="DECIMAL"/>
  12.         <result column="frozen" property="frozen" jdbcType="DECIMAL"/>
  13.     </resultMap>
  15.     <update id="decrease">
  16.         UPDATE account
  17.         SET residue = residue - #{money},
  18.             used    = used + #{money}
  19.         where user_id = #{userId};
  20.     </update>
  22.     <select id="findByUserId" resultMap="BaseResultMap">
  23.         select *
  24.         from account
  25.         where user_id = #{userId};
  26.     </select>
  28.     <update id="updateResidueToFrozen">
  29.         update account
  30.         set residue = residue - #{money},
  31.             frozen  = frozen + #{money}
  32.         where user_id = #{userId}
  33.     </update>
  35.     <update id="updateFrozenToUsed">
  36.         update account
  37.         set frozen  = frozen - #{money},
  38.             used = used + #{money}
  39.         where user_id = #{userId}
  40.     </update>
  42.     <update id="updateFrozenToResidue">
  43.         update account
  44.         set frozen  = frozen - #{money},
  45.             residue = residue + #{money}
  46.         where user_id = #{userId}
  47.     </update>
  49. </mapper>

添加幂等性处理机制

  1. public class ResultHolder {
  2.     private static Map<Class<?>, Map<String, String>> map = new ConcurrentHashMap<Class<?>, Map<String, String>>();
  4.     public static void setResult(Class<?> actionClass, String xid, String v) {
  5.         Map<String, String> results = map.get(actionClass);
  7.         if (results == null) {
  8.             synchronized (map) {
  9.                 if (results == null) {
  10.                     results = new ConcurrentHashMap<>();
  11.                     map.put(actionClass, results);
  12.                 }
  13.             }
  14.         }
  16.         results.put(xid, v);
  17.     }
  19.     public static String getResult(Class<?> actionClass, String xid) {
  20.         Map<String, String> results = map.get(actionClass);
  21.         if (results != null) {
  22.             return results.get(xid);
  23.         }
  25.         return null;
  26.     }
  28.     public static void removeResult(Class<?> actionClass, String xid) {
  29.         Map<String, String> results = map.get(actionClass);
  30.         if (results != null) {
  31.             results.remove(xid);
  32.         }
  33.     }
  35. }

添加TCC事务处理接口

  1. @LocalTCC
  2. public interface AccountTccAction {
  3.     @TwoPhaseBusinessAction(name = "AccountTccAction")
  4.     boolean prepare(BusinessActionContext context,
  5.                     @BusinessActionContextParameter(paramName = "userId") Long userId,
  6.                     @BusinessActionContextParameter(paramName = "money") BigDecimal money);
  8.     boolean commit(BusinessActionContext context);
  10.     boolean rollback(BusinessActionContext context);
  11. }

接口实现类

  1. @Component
  2. public class AccountTccActionImpl implements AccountTccAction {
  3.     private final AccountMapper accountMapper;
  5.     @Autowired
  6.     public AccountTccActionImpl(AccountMapper accountMapper) {
  7.         this.accountMapper = accountMapper;
  8.     }
  10.     @Transactional
  11.     @Override
  12.     public boolean prepare(BusinessActionContext context, Long userId, BigDecimal money) {
  13.         Account account = accountMapper.findByUserId(userId);
  14.         if (account.getResidue().compareTo(money) < 0) {
  15.             throw new RuntimeException("可用金额不足");
  16.         }
  17.         accountMapper.updateResidueToFrozen(userId, money);
  18.         ResultHolder.setResult(AccountTccAction.class, context.getXid(), "p");
  19.         return true;
  20.     }
  22.     @Transactional
  23.     @Override
  24.     public synchronized boolean commit(BusinessActionContext context) {
  25.         if (ResultHolder.getResult(AccountTccAction.class, context.getXid()) == null) {
  26.             return true;
  27.         }
  28.         Long userId = Long.valueOf(context.getActionContext("userId").toString());
  29.         BigDecimal money = new BigDecimal(context.getActionContext("money").toString());
  30.         accountMapper.updateFrozenToUsed(userId, money);
  31.         ResultHolder.removeResult(AccountTccAction.class, context.getXid());
  32.         return true;
  33.     }
  35.     @Transactional
  36.     @Override
  37.     public synchronized boolean rollback(BusinessActionContext context) {
  38.         if (ResultHolder.getResult(AccountTccAction.class, context.getXid()) == null) {
  39.             return true;
  40.         }
  41.         Long userId = Long.valueOf(context.getActionContext("userId").toString());
  42.         BigDecimal money = new BigDecimal(context.getActionContext("money").toString());
  43.         accountMapper.updateFrozenToResidue(userId, money);
  44.         ResultHolder.removeResult(AccountTccAction.class, context.getXid());
  45.         return true;
  46.     }
  47. }