使用springcloud-seata解决分布式事务问题-2PC模式

目录

一、建立undo_log表

二、安装事务协调器：seata-server

三、整合

可以查看官网：快速启动 | Seata 

一、建立undo_log表

-- 注意此处0.3.0+ 增加唯一索引 ux_undo_log
CREATE TABLE `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,
  `ext` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

二、安装事务协调器：seata-server

三、整合

（一） 导入依赖 spring-cloud-starter-alibaba-seata seata-all 0.7.1

（二） 启动seata-server（TC）

修改配置 registry.conf 修改seata的注册中心为nacos，（文件配置放在file.conf中，不修改）

registry.conf文件修改后如下：

registry {
  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa seata的注册中心
  type = "nacos"　　//修改的地方
 
  nacos {
    serverAddr = "localhost:8848"
    namespace = "public"
    cluster = "default"
  }
  eureka {
    serviceUrl = "http://localhost:1001/eureka"
    application = "default"
    weight = "1"
  }
  redis {
    serverAddr = "localhost:6379"
    db = "0"
  }
  zk {
    cluster = "default"
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
  }
  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
  type = "file"
 
  nacos {
    serverAddr = "localhost"
    namespace = "public"
    cluster = "default"
  }
  consul {
    serverAddr = "127.0.0.1:8500"
  }
  apollo {
    app.id = "seata-server"
    apollo.meta = "http://192.168.1.204:8801"
  }
  zk {
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
  }
  etcd3 {
    serverAddr = "http://localhost:2379"
  }
  file {
    name = "file.conf"
  }
}

（三） 所有想要用到分布式事务的微服务使用seata DatasourceProxy代理自己的数据源

@Autowired
DataSourceProperties dataSourceProperties;
​
@Bean
public DataSource dataSource(DataSourceProperties properties) {
    HikariDataSource dataSource = dataSourceProperties.initializeDataSourceBuilder().type(HikariDataSource.class).build();
    if (StringUtils.hasText(properties.getName())) {
        dataSource.setPoolName(properties.getName());
    }
    return new DataSourceProxy(dataSource);
}

（四） 每个微服务，都必须导入(放resource下)

register.conf和 file.conf(修改vgroup_mapping.{application.name}-fescar-service-group = "default")

（五） 在分布式事务的入口方法上设置@GlobalTransactional（TM），在每一小事务上设置@Transactional（RM）

远程调用的方法上面也要添加@Transactional

    @GlobalTransactional //seata分布式事务控制
    @Transactional
    @Override
    public SubmitOrderResponseVo submitOrder(OrderSubmitVo vo) {
 
        submitVoThreadLocal.set(vo);
        MemberResVo memberResVo = LoginUserInterceptor.loginUser.get();
 
        SubmitOrderResponseVo response = new SubmitOrderResponseVo();
        response.setCode(0);
        String redisToken = redisTemplate.opsForValue().get(OrderConstant.USER_ORDER_TOKEN_PREFIX + memberResVo.getId());
        String orderToken = vo.getOrderToken();
        // 成功返回1  失败返回0
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        // 保证原子性
        Long result = redisTemplate.execute(new DefaultRedisScript(script, Long.class), Arrays.asList(OrderConstant.USER_ORDER_TOKEN_PREFIX + memberResVo.getId()), orderToken);
        if(result == 0L) {
            // 验证失败
            response.setCode(1);
            return response;
        } else {
            // 下单，创建订单，校验令牌，检验价格，锁库存
            // TODO 1、创建订单，订单项等信息
            OrderCreateTo order = createOrder();
            // TODO 2、验价
            BigDecimal payAmount = order.getOrder().getPayAmount();
            if(Math.abs(payAmount.subtract(vo.getPayPrice()).doubleValue()) < 0.01) {
                // 金额对比成功后保存订单
                // TODO 3、保存订单
                saveOrder(order);
 
                WareSkuLockVo wareSkuLockVo = new WareSkuLockVo();
                wareSkuLockVo.setOrderSn(order.getOrder().getOrderSn());
                List collect = order.getOrderItems().stream().map(item -> {
                    OrderItemVo orderItemVo = new OrderItemVo();
                    orderItemVo.setCount(item.getSkuQuantity());
                    orderItemVo.setSkuId(item.getSkuId());
                    orderItemVo.setTitle(item.getSkuName());
                    return orderItemVo;
                }).collect(Collectors.toList());
                wareSkuLockVo.setLocks(collect);
                // TODO  4、锁库存
                // 出异常后，因为远程锁库存成功，但是忘了原因超时了，订单回滚，库存不回滚
                R r = wareFeignService.orderLockStock(wareSkuLockVo);
                if(r.getCode() == 0) {
                    // 锁成功
                    response.setOrder(order.getOrder());
 
                    // TODO 5 出异常
                    int i = 10/0;
                    return response;
                } else {
                    // 锁定失败
                    // 抛异常才能使事务回滚
                    response.setCode(3);
                    throw new NoStockException((String)r.get("msg"));
 
//                    return response;
                }
            } else {
                response.setCode(2); // 金额对比失败
                return response;
            }
 
        }

注意：

这种方法其实并不适用于下单等高并发场景。seata的AT模式适用于并发量不大的场景，比如说后台商家添加商品什么的，而不适用于高并发场景。因为锁太多导致串行，没法并发。