谷粒商城二十一订单服务分布式事务
Posted 阿无，
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了谷粒商城二十一订单服务分布式事务相关的知识，希望对你有一定的参考价值。
举个例子，订单服务的执行流程是，订单服务的逻辑、远程调用库存服务、远程调用优惠券服务。
事务不会出问题
-
假设订单服务出异常，代码都执行不到仓储服务锁库存、优惠券服务扣积分的远程调用步骤，不存在整体回滚的问题，事务不会出问题。
-
如果是远程调用的库存服务出异常，首先库存服务自己会回滚，订单服务会在库存服务调用完成之后获取返回的锁库存的结果，得知发生了异常，订单服务也会抛异常，那么订单服务也会回滚。
事务会出问题
-
假设我们订单服务是成功的，在调用库存服务的时候，出现了假失败的情况，例如服务器故障、超时等原因，库存锁成功并且库存服务事务提交了之后，一直没有给订单服务返回，远程调用有一个超时机制，远程调用会抛一个类似readTimeout的异常，就会出现一个问题，
库存锁成功了，订单却回滚了。 -
如果优惠券服务出异常，那么优惠券服务会自动回滚，订单服务获取到异常信息也会自动回滚，
但是库存服务不会回滚 -
如果订单服务在成功调用各远程服务，最后在整合结果的时候出现异常，那么
订单服务会回滚，远程服务则不会回滚
本地事务
数据库事务的几个特性：原子性(Atomicity )、一致性( Consistency )、隔离性或独立性( Isolation)
和持久性(Durabilily)，简称就是 ACID；
-
原子性：一系列的操作整体不可拆分，要么同时成功，要么同时失败
-
一致性：数据在事务的前后，业务整体一致。
转账。A:1000；B:1000； 转 200 事务成功; A：800 B：1200 -
隔离性：事务之间互相隔离。
例如100个人下单，就会有100个事务，有一个失败了，它的事务回滚不会影响其他事务。 -
持久性：一旦事务成功，数据一定会落盘在数据库。
事务的隔离级别
-
READ UNCOMMITTED（读未提交）
该隔离级别的事务会读到其它未提交事务的数据，此现象也称之为脏读。 -
READ COMMITTED（读提交）
一个事务可以读取另一个已提交的事务，多次读取会造成不一样的结果，此现象称为不可重复读问题，Oracle 和 SQL Server 的默认隔离级别。 -
REPEATABLE READ（可重复读）
该隔离级别是 mysql 默认的隔离级别，在同一个事务里，select 的结果是事务开始时时间点的状态，因此，同样的 select 操作读到的结果会是一致的，但是，会有幻读现象。MySQL的 InnoDB 引擎可以通过 next-key locks 机制（参考下文"行锁的算法"一节）来避免幻读。 -
SERIALIZABLE（序列化）
在该隔离级别下事务都是串行顺序执行的（也就没有并发能力了），MySQL 数据库的 InnoDB 引擎会给读操作隐式加一把读共享锁，从而避免了脏读、不可重读复读和幻读问题。
事务的传播行为和@Transactional的坑
PROPAGATION_REQUIRED：如果当前没有事务，就创建一个新事务，如果当前存在事务，就加入该事务，该设置是最常用的设置。- PROPAGATION_SUPPORTS：支持当前事务，如果当前存在事务，就加入该事务，如果当前不存在事务，就以非事务执行。
- PROPAGATION_MANDATORY：支持当前事务，如果当前存在事务，就加入该事务，如果当前不存在事务，就抛出异常。
PROPAGATION_REQUIRES_NEW：创建新事务，无论当前存不存在事务，都创建新事务。- PROPAGATION_NOT_SUPPORTED：以非事务方式执行操作，如果当前存在事务，就把当前事务挂起。
- PROPAGATION_NEVER：以非事务方式执行，如果当前存在事务，则抛出异常。
- PROPAGATION_NESTED：如果当前存在事务，则在嵌套事务内执行。如果当前没有事务，则执行与 PROPAGATION_REQUIRED 类似的操作。
/**
*
* 本地事务，在分布式系统，只能控制住自己的回滚，控制不了其它事务的回滚
* 分布式事务，最大原因，网络问题
*
* isolation = Isolation.SERIALIZABLE 事务的隔离级别
* propagation = Propagation.REQUIRED 事务的传播行为
* timeout = 30 事务的超时时间，该事务30s没有执行成功就回滚
*
*
* REQUIRED，REQUIRES_NEW
*
* 传播行为：
* a,b,c方法都标注了@Transactional，a调用了bc，bc事务是否要与a共用一个事务
* 如果此时b的传播行为是REQUIRED,c的传播行为是REQUIRES_NEW
* 那么b与a共用一个事务，c新建一个事务
* 此时如果a在调用bc之后发生了异常，则ab事务回滚，c不会回滚
*
* 如果ab事务都设置了timeout，那么b的timeout将失效
*
* springboot事务的坑：
* 如果abc方法都加了@Transactional，abc在同一个service类中，a调用了bc，那么bc任何事务的设置都无效
*
* 因为事务是用代理对象来控制的，如果像上面那样调用方法相当于把bc方法的代码直接粘贴到了a方法中
* 同一个对象内事务方法互调默认失效，原因 绕过了代理对象
*
* 解决：
* 使用代理对象来调用事务方法
* 1. 引入aop-stater模块，引入了aspectj
* 2. @EnableAspectJAutoProxy(exposeProxy = true)，开启aspectj动态代理功能。
*
* exposeProxy,对外暴露代理对象
*
* 以后所有的动态代理都是aspectj创建的（即使没有接口也可以创建动态代理）
* 默认是使用jdk接口的动态代理
*
* 3. 本类对象用代理对象来调用
* Object o = AopContext.currentProxy();
* 强转成我们的本类对象即可
*
*
*
* @param vo
* @return
*/
别人家有位在MIT读博士的小孩对我说，他们没有讲义，也没有课本；老师上来讲一通，黑板上写一通，指定一批参考读物；几名教授讲课时互相批判，各有各的观点，各自推荐一些参考书指导你去读；然后你写，写文章不需要结果，只要思维逻辑是合适的，老师就给你打分了，这就是科学。老师讲的是他正在研究的前沿，他自己也未完全明白，科学是一代一代去努力的。博士只要有思想方法，不一定会有结果，有些科学太遥远，一步很难证明，过程正确也是好的。我们火花奖也不一定要有“结果”才行，否则怎么叫“青出于蓝而胜于蓝”呢？如果教育总是追求结果，学生思想就会被约束。所以，不能太实用主义，以需求为导向，牵引中国前进是不够的。教育应该放开让学生“胡思乱想”，只要他想的逻辑相吻合，就不要去约束他。读书是为了拿到开门的“钥匙”，关键是读一个方法学，运用知识的方法比知识更重要。因为知识可以在互联网上获取，怎么组合、怎么拼接，这就是大学要学习的。
https://baijiahao.baidu.com/s?id=1760664270073856317&wfr=spider&for=pc
擦亮花火、共创未来——任正非在“难题揭榜”花火奖座谈会上的讲话
任正非
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.以上是关于谷粒商城二十一订单服务分布式事务的主要内容，如果未能解决你的问题，请参考以下文章