Day434.订单&库存服务分布式事务的最终解决 -谷粒商城
Posted 阿昌喜欢吃黄桃
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Day434.订单&库存服务分布式事务的最终解决 -谷粒商城相关的知识，希望对你有一定的参考价值。
分布式事务的最终解决
通过MQ的延时消息来解决订单/库存服务直接调用的事务问题
一、如何解决下单系统这个高并发里边的分布式事务呢？
首先，我们肯定不会用 2PC 模式、 TCC-事务补偿性方案，我们也不考虑
最终我们选择可靠消息+最终一致性这种方式
为了保证高并发，订单这一块还是自己回滚，
因为性能不佳，我们弃用seata的方案
1、库存服务自己怎么回滚？
- 第一种
通过MQ消息队列，我们在提交订单那里，当捕捉到异常要回滚的时候，给库存服务发一个消息，让库存服务自己把库存解锁
这样不需要让库存事务回滚，只需要给它发一个消息，不会损失什么性能
- 第二种
通过MQ延迟消息
库存服务本身也可以使用自动解锁模式。
怎么自动解锁呢？
需要使用消息队列来完成。
如果你想让我这的哪些库存解锁，首先你需要给我发一个消息告诉我。
然后我们专门的库存解锁服务，去来订阅我们stock.release.stock.queue这个队列里的消息。
那你给我发消息的时候，比如：用路由键stock.release，我知道要库存解锁，
然后，你的消息发给我们这个交换机stock-event-exchange。
交换机把这个消息路由给stock.release.stock.queue这个队列。
然后，这个队列stock.release.stock.queue里边存的这些消息都是库存要解锁的消息，我们的库存解锁服务只要收到了，它就会在后台慢慢的解锁消息。
我们不用保证强一致，我们哪怕是二十分钟、三十分钟，乃至于一天以后把这个库存解锁了，最终一致了就行。
所以我们可以来使用消息队列来完成我们的这个最终一致性。
2、锁库存的增强版逻辑
我们想要锁库存的话，我们先来保存一个库存工作单和库存工作单详情
相当于只要我们想要锁库存，我们先给数据库里边保存记录，我要锁库存。
接下来我们就来进行锁，只要锁成功了，那一切ok。
如果锁失败了，数据库里边相当于没有这个锁库存记录。
因为锁失败呢，我们这个本身自己所失败会全部回滚。
但如果可能是这种失败，比如我们来到订单里边，我们库存其实自己锁成功了。但是我们订单下边的其他完了，然后库存要进行解锁。那怎么办呢？
我们可以使用定时任务
二、订单服务的完整消息队列
1、库存自动解锁
库存微服务，有一个它的库存交换机stock-event-exchange.
如果想要解锁库存，应该是这样的。
首先订单创建成功之后，库存锁定成功，然后发一个消息给交换机，
这个消息里面的内容有订单编号、仓库编号、哪个商品锁了几个库存，
这个交换机，绑定了两个队列，
一个是按照stock.release.#模糊匹配的路由键绑定的stock.release.stock.queue队列
一个是stock.delay.queue队列
第一次发的库存锁定成功的消息，先使用路由键叫stock.locked
交换机按照这个路由键，找到stock.delay.queue延时队列
延时队列50分钟以后，用stock.release这个路由键，将死信交给库存交换机stock-event-exchange，
交换机收到以后，按照这个路由键查找，发现stock.release.#这个模糊匹配的路由键跟它是一样的，然后被交换机路由到我们这个stock.release.stock.queue队列。
接下来的解锁库存服务，专门来处理stock.release.stock.queue里的消息。
- 最终实现
- com.achang.achangmall.order.conf.MQConfig
通过配置类来创建队列、绑定、交换机
@Configuration
public class MQConfig {
@Bean
public Queue orderDelayQueue(){//死信队列
HashMap<String, Object> argument = new HashMap<>();
argument.put("x-dead-letter-exchange","order-event-exchange");//死信路由
argument.put("x-dead-letter-routing-key","order.release.order");//死信路由键
argument.put("x-message-ttl",60000);//消息过期时间 1分钟
return new Queue("order.delay.queue",true,false,false,argument);
}
@Bean
public Queue orderReleaseOrderQueue(){//普通队列
return new Queue("order.release.order.queue",true,false,false,null);
}
@Bean
public Exchange orderEventExchange(){//普通的主题交换机（可绑定多个队列，根据条件路由）
return new TopicExchange("order-event-exchange",true,false);
}
@Bean
public Binding orderCreateOrderBinding(){
return new Binding("order.delay.queue",
Binding.DestinationType.QUEUE,
"order-event-exchange",
"order.create.order",null);
}
@Bean
public Binding orderReleaseOrderBinding(){
return new Binding("order.release.order.queue",
Binding.DestinationType.QUEUE,
"order-event-exchange",
"order.release.order",null);
}
}
- 库存服务加mq依赖
achangmall-ware/pom.xml
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
- rabbitmq配置
achangmall-ware/src/main/resources/application.yaml
spring:
rabbitmq:
host: 192.168.109.101
port: 5672
virtual-host: /
listener:
simple:
acknowledge-mode: manual
- 主启动类启动MQ
achangmall-ware/src/main/java/com/achang/achangmall/AchangmallWareApplication.java
@EnableRabbit
- 配置类
com.achang.achangmall.ware.conf.RabbitMQConfig
@Configuration
public class RabbitMQConfig {
@Bean
public MessageConverter messageConversionException(){
return new Jackson2JsonMessageConverter();
}
@Bean
public Exchange stockEventExchange(){
return new TopicExchange("stock-event-exchange",true,false);
}
@Bean
public Queue stockReleaseStockQueue(){
return new Queue("stock.release.stock.queue",true,false,false);
}
@Bean
public Queue stockDelayQueue(){
HashMap<String, Object> argument = new HashMap<>();
argument.put("x-dead-letter-exchange","stock-event-exchange");//死信路由
argument.put("x-dead-letter-routing-key","stock.release");//死信路由键
argument.put("x-message-ttl", 120000L);//消息过期时间 1分钟
return new Queue("stock.delay.queue",true,false,false,argument);
}
@Bean
public Binding stockReleaseBinding(){
return new Binding("stock.release.stock.queue",
Binding.DestinationType.QUEUE,
"stock-event-exchange",
"stock.release.#",
null);
}
@Bean
public Binding stockLockedBinding(){
return new Binding("stock.delay.queue",
Binding.DestinationType.QUEUE,
"stock-event-exchange",
"stock.locked",
null);
}
}
- com.achang.achangmall.ware.controller.WareSkuController
/**
* 锁定库存
* @param vo
*
* 库存解锁的场景
* 1）、下订单成功，订单过期没有支付被系统自动取消或者被用户手动取消，都要解锁库存
* 2）、下订单成功，库存锁定成功，接下来的业务调用失败，导致订单回滚。之前锁定的库存就要自动解锁
* 3）、
*/
@PostMapping(value = "/lock/order")
public R orderLockStock(@RequestBody WareSkuLockVo vo) {
// try {
boolean lockStock = wareSkuService.orderLockStock(vo);
return R.ok().setData(lockStock);
// } catch (NoStockException e) {
// return R.error(NO_STOCK_EXCEPTION.getCode(),NO_STOCK_EXCEPTION.getMessage());
// }
}
- 发送到mq消息队列的to
com.achang.common.to.mq.StockLockedTo
/**
* @Description: 发送到mq消息队列的to
**/
@Data
public class StockLockedTo {
/** 库存工作单的id **/
private Long id;
/** 工作单详情的所有信息 **/
private StockDetailTo detailTo;
}
- com.achang.common.to.mq.StockDetailTo
@Data
public class StockDetailTo {
private Long id;
/**
* sku_id
*/
private Long skuId;
/**
* sku_name
*/
private String skuName;
/**
* 购买个数
*/
private Integer skuNum;
/**
* 工作单id
*/
private Long taskId;
/**
* 仓库id
*/
private Long wareId;
/**
* 锁定状态
*/
private Integer lockStatus;
}
- com.achang.achangmall.ware.service.impl.WareSkuServiceImpl
@Service("wareSkuService")
public class WareSkuServiceImpl extends ServiceImpl<WareSkuDao, WareSkuEntity> implements WareSkuService {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private WareSkuDao wareSkuDao;
@Autowired
private OrderFeignService orderFeignService;
@Autowired
private WareOrderTaskDetailService wareOrderTaskDetailService;
@Override
public PageUtils queryPage(Map<String, Object> params) {
QueryWrapper<WareSkuEntity> wrapper = new QueryWrapper<>();
String skuId = (String) params.get("skuId");
if (!StringUtils.isEmpty(skuId)){
wrapper.eq("sku_id",skuId);
}
String wareId = (String) params.get("wareId");
if (!StringUtils.isEmpty(wareId)){
}
IPage<WareSkuEntity> page = this.page(
new Query<WareSkuEntity>().getPage(params),
wrapper
);
return new PageUtils(page);
}
@Override
public List<SkuHasStockVo> getSkuHasStock(List<Long> skuIdList) {
List<SkuHasStockVo> collect = skuIdList.stream().map(item -> {
SkuHasStockVo vo = new SkuHasStockVo();
Long count = baseMapper.getSkuStock(item);
vo.setSkuId(item);
vo.setHasStock(count == null?false:count>0);
return vo;
}).collect(Collectors.toList());
return collect;
}
@Autowired
private WareOrderTaskService wareOrderTaskService;
@Override
public boolean orderLockStock(WareSkuLockVo vo) {
/**
* 保存库存工作单详情信息
* 追溯
*/
WareOrderTaskEntity wareOrderTaskEntity = new WareOrderTaskEntity();
wareOrderTaskEntity.setOrderSn(vo.getOrderSn());
wareOrderTaskEntity.setCreateTime(new Date());
wareOrderTaskService.save(wareOrderTaskEntity);
//1、按照下单的收货地址，找到一个就近仓库，锁定库存
//2、找到每个商品在哪个仓库都有库存
List<OrderItemVo> locks = vo.getLocks();
List<SkuWareHasStock> collect = locks.stream().map((item) -> {
SkuWareHasStock stock = new SkuWareHasStock();
Long skuId = item.getSkuId();
stock.setSkuId(skuId);
stock.setNum(item.getCount());
//查询这个商品在哪个仓库有库存
List<Long> wareIdList = wareSkuDao.listWareIdHasSkuStock(skuId);
stock.setWareId(wareIdList);
return stock;
}).collect(Collectors.toList());
//2、锁定库存
for (SkuWareHasStock hasStock : collect) {
boolean skuStocked = false;
Long skuId = hasStock.getSkuId();
List<Long> wareIds = hasStock.getWareId();
if (org.springframework.util.StringUtils.isEmpty(wareIds)) {
//没有任何仓库有这个商品的库存
// throw new NoStockException(skuId);
}
//1、如果每一个商品都锁定成功,将当前商品锁定了几件的工作单记录发给MQ
//2、锁定失败。前面保存的工作单信息都回滚了。发送出去的消息，即使要解锁库存，由于在数据库查不到指定的id，所有就不用解锁
for (Long wareId : wareIds) {
//锁定成功就返回1，失败就返回0
Long count = wareSkuDao.lockSkuStock(skuId,wareId,hasStock.getNum());
if (count == 1) {
skuStocked = true;
WareOrderTaskDetailEntity taskDetailEntity = new WareOrderTaskDetailEntity();
taskDetailEntity.setSkuId(skuId);
taskDetailEntity.setSkuName("");
taskDetailEntity.setSkuNum(hasStock.getNum());
taskDetailEntity.setTaskId(wareOrderTaskEntity.getId());
taskDetailEntity.setWareId(wareId);
taskDetailEntity.setLockStatus(1);
wareOrderTaskDetailService.save(taskDetailEntity);
//TODO 告诉MQ库存锁定成功
StockLockedTo lockedTo = new StockLockedTo();
lockedTo.setId(wareOrderTaskEntity.getId());
StockDetailTo detailTo = new StockDetailTo();
BeanUtils.copyProperties(taskDetailEntity,detailTo);
lockedTo.setDetailTo(detailTo);
rabbitTemplate.convertAndSend("stock-event-exchange","stock.locked",lockedTo);
break;
} else {
//当前仓库锁失败，重试下一个仓库
}
}
if (skuStocked == false) {
//当前商品所有仓库都没有锁住
throw new NoStockException(skuId);
}
}
//3、肯定全部都是锁定成功的
return true;
}
/**
* 解锁库存的方法
* @param skuId
* @param wareId
* @param num
* @param taskDetailId
*/
public void unLockStock(Long skuId,Long wareId,Integer num,Long taskDetailId) {
//库存解锁
wareSkuDao.unLockStock(skuId,wareId,num);
//更新工作单的状态
WareOrderTaskDetailEntity taskDetailEntity = new WareOrderTaskDetailEntity();
taskDetailEntity.setId(taskDetailId);
//变为已解锁
taskDetailEntity.setLockStatus(2);
wareOrderTaskDetailService.updateById(taskDetailEntity);
}
@Data
class SkuWareHasStock {
private Long skuId;
private Integer num;
private List<Long> wareId;
}
/**
* 防止订单服务卡顿，导致订单状态消息一直改不了，库存优先到期，查订单状态新建，什么都不处理
* 导致卡顿的订单，永远都不能解锁库存
* @param orderTo
*/
@Transactional(rollbackFor = Exception.class)
@Override
public void unlockStock(OrderTo orderTo) {
String orderSn = orderTo.getOrderSn();
//查一下最新的库存解锁状态，防止重复解锁库存
WareOrderTaskEntity orderTaskEntity = wareOrderTaskService.getOrderTaskByOrderSn(orderSn);
//按照工作单的id找到所有 没有解锁的库存，进行解锁
Long id = orderTaskEntity.getId();
List<WareOrderTaskDetailEntity> list = wareOrderTaskDetailService.list(new QueryWrapper<WareOrderTaskDetailEntity>()
.eq("task_id", id).eq("lock_status", 1));
for (WareOrderTaskDetailEntity taskDetailEntity : list) {
unLockStock(taskDetailEntity.getSkuId(),
taskDetailEntity.getWareId(),
taskDetailEntity.以上是关于Day434.订单&库存服务分布式事务的最终解决 -谷粒商城的主要内容，如果未能解决你的问题，请参考以下文章