SpringCloud分布式框架&分布式事务&分布式锁
总结
本文承接上一篇SpringCloud分布式框架实践之后，进一步实践分布式事务与分布式锁，其中分布式事务主要是基于Seata的AT模式进行强一致性，基于RocketMQ事务消息进行最终一致性，分布式锁直接使用Redisson进行实践。
分布式事务
Seata分布式事务
Seata分布式事务支持很多模式，例如XA模式、TCC模式、AT模式和SAGA模式，其中AT模式是默认模式，也是无业务入侵的最简单集成的模式。这里我们只针对AT模式进行实践，其他模式Seata官网相关原理技术讲的很清楚，可自行实践 快速启动 | Apache Seata。
- 下载Seata安装包 「链接」，选择自己需要的版本，下载之后直接解压进入“\seata-server\bin”目录，cmd执行seata-server.bat即可。可以在“seata-server\conf”目录下application.yml修改相关配置，比如将相关配置信息注册到nacos等，可以参考同步录下的example文件，这里我直接单机启动。启动之后，控制台访问地址http://127.0.0.1:7090，账号密码seata
- 启动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的消息事务最终一致性的方案也是一种不错的分布式事务解决方案。
- 首先我们先下载RocketMQ「链接」，将下载的文件进行解压，在“/bin”目录下基于cmd运行“mqnamesrv.cmd”和“mqbroker.cmd”，
- 如果想要访问控制台，需要再单独下载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端口，如果冲突可以进行修改。
- 生产者和消费者进行集成，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全部解决。
- POM引入依赖
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.17.7</version>
</dependency>- 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, "参数不可以为空");
}