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

ゝ一纸荒年。 2023-10-16 08:44 45阅读 0赞

**目录**

一、建立undo\_log表

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

三、整合

--------------------

可以查看官网：[快速启动 | Seata][_ 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下)**

**（五） 在分布式事务的入口方法上设置@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<Long>(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<OrderItemVo> 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模式适用于并发量不大的场景，比如说后台商家添加商品什么的，而**不适用于高并发场景。**因为锁太多导致串行，没法并发。

[_ Seata]: http://seata.io/zh-cn/docs/user/quickstart/