总结

本文承接上一篇SpringCloud分布式框架实践之后，进一步实践分布式事务与分布式锁，其中分布式事务主要是基于Seata的AT模式进行强一致性，基于RocketMQ事务消息进行最终一致性，分布式锁直接使用Redisson进行实践。

分布式事务

Seata分布式事务

Seata分布式事务支持很多模式，例如XA模式、TCC模式、AT模式和SAGA模式，其中AT模式是默认模式，也是无业务入侵的最简单集成的模式。这里我们只针对AT模式进行实践，其他模式Seata官网相关原理技术讲的很清楚，可自行实践 快速启动 | Apache Seata。

1. 下载Seata安装包 「链接」，选择自己需要的版本，下载之后直接解压进入“\seata-server\bin”目录，cmd执行seata-server.bat即可。可以在“seata-server\conf”目录下application.yml修改相关配置，比如将相关配置信息注册到nacos等，可以参考同步录下的example文件，这里我直接单机启动。启动之后，控制台访问地址http://127.0.0.1:7090，账号密码seata

2. 启动Seata服务端之后，这里我们基于Springboot集成Seata客户端，首先我们需要在涉及分布式事务的数据库添加一个undo_log表，然后需要引入依赖，进行服务端配置，最后添加上事务注解，即可支持分布式事务。

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;

POM依赖引入：

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

application.yml配置：

# 这里由于我们是直接部署，未将seata信息注入到nacos，故需要进行单机配置
seata:
  enabled: true
  application-id: seata-server
  tx-service-group: seata-test-group
  service:
    vgroup-mapping:
      seata-test-group: default
    grouplist:
      default: 127.0.0.1:8091
  registry:
    type: file
  config:
    type: file

在需要添加分布式事务的方法上添加事务注解@GlobalTransactional

    @GlobalTransactional
    public Long save(OrderInfo orderInfo) {
        OrderEntity orderEntity = new OrderEntity();
        orderEntity.setOrderNo(orderInfo.getOrderNo());
        orderEntity.setTotalPrice(orderInfo.getTotalPrice());
        this.save(orderEntity);
      	// 这里远程调用其他服务
        productAPI.updateProduct(orderInfo.getProductList().get(0));
        return orderEntity.getOrderNo();
    }

    @Override
    public ProductInfo updateProduct(ProductInfo productInfo) {
        ProductEntity byId = getById(productInfo.getProductId());
        byId.setProductNum(byId.getProductNum() - 1);
        updateById(byId);
        int i = 1 / 0;
        return productInfo;
    }

RocketMQ消息事务

由于强一致性的事务，一般涉及到资源抢夺加锁和数据阻塞导致性能不佳，故基于RocketMQ的消息事务最终一致性的方案也是一种不错的分布式事务解决方案。

1. 首先我们先下载RocketMQ「链接」，将下载的文件进行解压，在“/bin”目录下基于cmd运行“mqnamesrv.cmd”和“mqbroker.cmd”，

2. 如果想要访问控制台，需要再单独下载GitHub - apache/rocketmq-dashboard at rocketmq-dashboard-2.0.0 源码，下载完成之后，直接在pom目录下打开Powershell窗口，执行mvn clean install -Dmaven.test.skip=true完成打包，然后进入target目录，执行“java -jar rocketmq-dashboard-版本号-SNAPSHOT --server.port=8081”进行启动，这里默认使用的是8080端口，如果冲突可以进行修改。

3. 生产者和消费者进行集成，POM引入依赖，添加配置文件，以及编写发送消息与消费消息代码。

POM依赖：

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.2.0</version>
        </dependency>

application.yml添加服务器配置：

# 生产者配置文件添加配置
rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: order-group

# 消费者添加配置
rocketmq:
  name-server: 127.0.0.1:9876

生产者消费消息：

@Slf4j
@Component
@RocketMQMessageListener(topic = "order-topic", consumerGroup = "product-group")
public class RocketMQConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String s) {
        // 生产者保证消息发送之后，并不关注消费者的消费是否成功，故消费者需要做好幂等或是补偿处理
        log.info("consumer accept : {}", s);
    }
}

生产者生产消息以及业务处理，由于RocketMQ需要与本地的数据处理保证原子性，故在发送消息到服务器上后，消息还只是半提交状态，用户不可消费。然后RocketMQ需要进行回调生产者，由生产者告诉RocketMQ是否开放消息状态。

    // 这里是模拟下订单之后，减产品库存
    public String msg(OrderInfo orderInfo) {
        String transactionId = UUID.randomUUID().toString();
        // 这里拿订单id作为了整个事务的全局标识，发送减库存消息。不处理订单业务
        rocketMQProducer.sendMsgInTransaction(JSON.toJSONString(orderInfo.getProductList().get(0)), transactionId);
        return transactionId;
    }
    // 发送消息
    public void sendMsgInTransaction(String message, String transactionId) {
        rocketMQTemplate.sendMessageInTransaction("order-topic", MessageBuilder.withPayload(message).setHeader("transactionId", transactionId).build(), null);
    }


/********************* RocketMQ消息回调处理 *******************************/
@RocketMQTransactionListener(rocketMQTemplateBeanName = "rocketMQTemplate")
@Slf4j
public class OrderMsgTransactionalListener implements RocketMQLocalTransactionListener {

    @Autowired
    private OrderTransactionalService orderTransactionalService;

    @Autowired
    private OrderService orderService;

    @Override
    // RocketMQ收到消息之后，会进行回调该接口，再该接口完成订单业务的处理，并返回RocketMQ提交消息还是回滚消息
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        MessageHeaders headers = msg.getHeaders();
        String transactionId = headers.get("transactionId", String.class);
        try {
            OrderEntity orderEntity = new OrderEntity();
            orderEntity.setOrderNo(transactionId);
            // 这里我们提交订单数据
            boolean b = orderService.commitOrder(orderEntity);
            log.info("订单 {} 事务提交结果 {} ", transactionId, b);
            // 返回事务处理结果
            return b ? RocketMQLocalTransactionState.COMMIT : RocketMQLocalTransactionState.ROLLBACK;
        } catch (Exception e) {
            log.error("订单 {} 事务提交失败", transactionId, e);
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }

    @Override
    // 如果executeLocalTransaction方法未立刻返回结果，RocketMQ会周期调用该方法进行事务状态判定，如果调用不通，则回滚消息。
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        Long transactionId = msg.getHeaders().get("transactionId", Long.class);
        if (orderTransactionalService.getById(transactionId) != null) {
            log.info("订单 {} 事务回查结果 {} ", transactionId, "COMMIT");
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            log.info("订单 {} 事务回查结果 {} ", transactionId, "ROLLBACK");
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}

分布式锁

分布式场景下常用的分布式锁，以前自己结合redis实现时，总是碰到诸多问题，例如锁超时自动释放但是业务未处理完，再就是锁重入问题等。这里集成redisson全部解决。

1. POM引入依赖

        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson-spring-boot-starter</artifactId>
            <version>3.17.7</version>
        </dependency>

2. Redisson集成与使用

// 添加配置项，初始化客户端
@Configuration
public class RedissonConfig {
    @Value("${spring.redis.host}")
    private String redisHost;
    @Value("${spring.redis.port}")
    private int redisPort;

    @Bean
    RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://" + redisHost + ":" + redisPort);
        return Redisson.create();
    }

}

    // 这里以账号注册为例，其中用户名不可以重复
    @Override
    public Result<Long> register(UserInfo userInfo) {
        UserInfoEntity userInfoEntity = new UserInfoEntity();
        if (userInfo.getUsername() != null && userInfo.getPassword() != null) {
            String lockName = userInfo.getUsername() + userInfo.getPassword();
            // 这里获取redis锁，锁在redis会以lockName作为key，value为hash的数据，进行存储
            RLock lock = redissonClient.getLock(lockName);
            boolean locked = false;
            try {
                // 尝试获取锁，这里只会有一个线程能拿到锁
                locked = lock.tryLock();
                if (locked) {
                    RBucket<String> bucket = redissonClient.getBucket(userInfo.getUsername());
                    if (bucket.isExists()) {
                        return Result.failed(null, "当前用户名已存在");
                    } else {
                        QueryWrapper<UserInfoEntity> queryWrapper = new QueryWrapper<>();
                        queryWrapper.eq("username", userInfo.getUsername());
                        UserInfoEntity one = getOne(queryWrapper);
                        if (one == null) {
                            userInfoEntity.setUsername(userInfo.getUsername());
                            userInfoEntity.setPassword(userInfo.getPassword());
                            save(userInfoEntity);
                            return Result.success(userInfoEntity.getId());
                        } else {
                            // 将查询到的用户名放入redis中
                            bucket.set(userInfo.getUsername());
                            return Result.failed(null, "当前用户名已存在");
                        }
                    }
                } else {
                    return Result.failed(null, "当前用户名正在注册中");
                }
            } finally {
                if (locked) {
                    try {
                        // 这里进行睡眠，不进行锁释放，redisson会另起一个线程对锁进行续期
                        Thread.sleep(50000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    lock.unlock();
                }
            }
        }
        return Result.failed(null, "参数不可以为空");
    }

一叶障目不见泰山