RabbitMQ（五）延时队列
延时队列的使用场景：
- 未支付订单，超过一段时间后，系统自动取消订单并释放占有物品
- 锁定库存一段时间后，检查订单不存在或者被取消，则解锁库存
1 定时任务存在的问题
如果使用Spring Schedule定时轮询数据库，则
-
消耗系统内存
-
增加数据库的压力
-
存在较大时间误差
存在时间误差的意思，是比如在定时器第一次扫描数据库后进行了订单的创建，然后第二次扫描就并不会扫描到该记录，只有等到第三次才能够扫描到
2 延时队列的实现
- 延时队列即是
RabbitMQ的消息TTL和死信Exchange的结合 消息的TTL（Time To Live）就是消息的存活时间，RabbitMQ可以对队列和消息分别设置TTL- 对队列的设置就是队列没有
消费者连接的保留时间，也可以是对每一个单独的消息做单独的设置，消息超过了这个时间还没有被消费，就认为消息死了，称之为死信 - 如果队列和消息同时设置了TTL，则会取最小值。因此如果一个消息被路由到不同的队列，那么消息死亡的时间有可能不一样（不同队列的TTL不同）
- 消息的TTL可以通过
expiration字段和x-message-ttl属性来设置过期时间，两者的效果是一样的
- 对队列的设置就是队列没有
2.1 死信与死信路由
-
死信（Dead Letter Exchanges，DLX）是指进入死信路由的消息，通过控制消息在一段时间后变为死信，然后控制变为死信的消息路由到指定的交换机（死信路由）来自己实现延时队列 -
一个消息满足下面的条件，就会进入
死信路由成为死信：- 一个消息被Consumer拒收了，并且reject方法中的参数requeue是false（不重新放回队列被其他消费者消费）
- 消息的TTL到期
- 队列的长度限制满了，排在前面的消息会被丢弃或者扔到
死信路由上
-
死信路由（Dead Letter Exchange）其实就是普通的exchange，只是设置了死信路由（Dead Letter Exchange）的队列会在队列内的消息过期的时候，自动触发消息的转发将其发给绑定的死信路由注意
对于延时队列，客户端监需要听的队列不是绑定死信路由的队列！！！而是变成死信后放入死信路由又被路由到的队列，从而保证从该队列取到的消息都是TTL过期的死信，这样就实现了延时队列的效果
2.2 通过队列TTL实现延时队列
- 首先
消息发送者（publisher）向消息队列发送消息，并指定路由键routing-key 交换机收到消息后，根据指定的路由键交给绑定的队列- 绑定的队列比较特殊，有以下几种设置：
x-message-ttl：队列内消息的存活时间x-dead-letter-routing-key：指定的死信路由的路由键x-dead-letter-exchange：指定的死信路由
死信路由收到过期的消息后，根据指定的路由键再去将消息发送给相应队列- 这样第二个队列内部的消息就都是超过TTL的消息了
2.3 通过消息TTL实现延时队列
- 首先
消息发送者（publisher）向消息队列发送消息，并指定路由键routing-key和消息的过期时间TTL - 消息队列等待队首等待时间过期自动将消息发送给路由键绑定的
交换机，交换机收到消息后，根据指定的路由键交给绑定的队列 - 绑定的队列比较特殊，有以下几种设置：
x-dead-letter-routing-key：指定的死信路由的路由键x-dead-letter-exchange：指定的死信路由
死信路由收到过期的消息后，根据指定的路由键再去将消息发送给相应队列- 这样第二个队列内部的消息就都是超过TTL的消息了
使用基于消息TTL的延时队列存在一些问题：
- 由于队列先进先出的特性，如果队首的过期时间比较长，就会导致队列的堵塞，后面TTL时间已经过期的消息就不能够及时放入
- 因此一般采用基于队列TTL的方式实现消息队列
3 延时队列场景1：实现自动关单
基于消息TTL的方式实现延时队列：
将两个交换机复用为一个，则得到改进版，其订单自动关闭的过程为：
- 订单创建成功后，订单服务作为消息的发送者（publisher）将消息发送给交换机
order-event-exchange，交换机按照其指定的路由键order.create.order（orderCreateBinding）发送给路由键绑定的队列order.delay.queue(orderDelayQueue) - 队列
order.delay.queue(orderDelayQueue)是一个延迟队列，设置的参数有：x-dead-letter-exchange：死信后交递的路由对象，这里的值为order-event-exchangex-dead-letter-route-key：死信后交递消息使用的路由键，这里的值为order-event-exchangex-message-ttl：队列的TTL
- 消息在延时队列成为死信后，会设置
死信路由键(值为orderReleaseBinding)并发送给死信交换器（复用了前面的交换器order-event-exchange），交换器收到消息后，按照消息的死信路由键(值为orderReleaseBinding)将其发送给绑定的订单释放队列(值为orderReleaseQueue)中 - 这样
订单释放队列(值为orderReleaseQueue)中存放的就是超过TTL时间的订单了
3.1 创建延时队列
- 创建延时队列、订单释放队列以及相应的交换器和路由键，方法是在一个配置类中通过
@Bean添加到容器，容器启动的时候就会自动连接到RabbitMQ，如果不存在就创建这些队列、交换器和路由键（如果存在就不会进行任何处理） - 这种方法还必须要有对应队列的Listener监听者才能创建成功，详情参考下面的测试
@Configuration
public class MQConfig {
/**
* 如果RabbitMQ中没有，则创建这些队列、交换器和路由键
* @return Queue
*/
@Bean
public Queue orderDelayQueue() {
Map<String, Object> args = new HashMap<>();
// 指定死信交换器
args.put("x-dead-letter-exchange", "order-event-exchange");
// 指定死信路由键
args.put("x-dead-letter-routing-key", "order.release.order");
// 指定队列TTL
args.put("x-message-ttl", 60000);
return new Queue("order.delay.queue", true, false, false, args);
}
/**
* 订单释放队列
* @return Queue
*/
@Bean
public Queue orderReleaseQueue() {
return new Queue("order.release.queue", true, false, false);
}
/**
* topic交换机
* @return TopicExchange
*/
@Bean
public Exchange orderEventExchange() {
return new TopicExchange("order-event-exchange", true, false);
}
/**
* 交换机和死信队列的绑定关系
* @return Binding
*/
@Bean
public Binding orderCreateBinding() {
return new Binding("order.delay.queue",
Binding.DestinationType.QUEUE,
"order-event-exchange",
"order.create.order",
null);
}
/**
* 交换机和订单释放队列的绑定关系
* @return Binding
*/
@Bean
public Binding orderReleaseBinding() {
return new Binding("order.release.queue",
Binding.DestinationType.QUEUE,
"order-event-exchange",
"order.release.order",
null);
}
}
3.2 测试
-
创建一个队列监听者
@RabbitListener(queues = "order.release.queue") public void listener(OrderEntity orderEntity, Channel channel, Message message) throws IOException { channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); System.out.println("收到过期的订单信息，准备关闭订单！" + orderEntity.toString()); }由于在之前配置中设置的手动确认，因此上面必须ack才能够将消息从release队列中移除
rabbitmq: # 开启发送端消息抵达队列的回调 publisher-returns: true # 抵达队列后以异步方式优先回调 template: mandatory: true # 开启发送端消息抵达broker的通知 publisher-confirm-type: correlated listener: simple: acknowledge-mode: manual -
编写一个测试controller，模拟提交订单
@RestController public class testController { @Autowired RabbitTemplate rabbitTemplate; @GetMapping("/test/submitOrder") public String submitOrderTest() { OrderEntity orderEntity = new OrderEntity(); orderEntity.setOrderSn(UUID.randomUUID().toString()); orderEntity.setModifyTime(new Date()); rabbitTemplate.convertAndSend("order-event-exchange", "order.create.order", orderEntity); return "ok"; } } -
下面可以看到，消息先被发送到了延时队列
order.delay.queue
-
在一分钟成为死信之后，由监听者从队列
order.release.queue中取出，并在控制台打印输出：
收到过期的订单信息，准备关闭订单！OrderEntity(id=null, memberId=null, orderSn=23564129-7557-4297-97d5-1a54a63c38df, couponId=null, createTime=null, memberUsername=null, totalAmount=null, payAmount=null, freightAmount=null, promotionAmount=null, integrationAmount=null, couponAmount=null, discountAmount=null, payType=null, sourceType=null, status=null, deliveryCompany=null, deliverySn=null, autoConfirmDay=null, integration=null, growth=null, billType=null, billHeader=null, billContent=null, billReceiverPhone=null, billReceiverEmail=null, receiverName=null, receiverPhone=null, receiverPostCode=null, receiverProvince=null, receiverCity=null, receiverRegion=null, receiverDetailAddress=null, note=null, confirmStatus=null, deleteStatus=null, useIntegration=null, paymentTime=null, deliveryTime=null, receiveTime=null, commentTime=null, modifyTime=Tue Aug 01 17:46:22 CST 2023)
3.3 自动关单完成
3.4 消息丢失、积压、重复等解决方案
4 延时队列场景2：自动库存解锁
4.1 整体流程
-
订单创建成功后，库存服务进行库存锁定，锁定成功后库存服务将作为消息的发送者
publisher向交换器发送发消息，并指定路由键为stock.locked -
交换器在收到消息后，将消息根据
路由键路由到延迟队列，等到消息在延时队列中超时成为死信后，再自动设置路由键位stock.release发往交换机 -
交换机收到死信后，根据
路由键进行模糊匹配，将消息发送给库存解锁队列不是很明白这里为什么要用模糊匹配？
-
这样监听库存解锁队列的库存解锁服务所收到的消息，都是锁定库存时间超时的消息了
4.2 创建延时队列
-
引入AMQP依赖
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> -
配置rabbirMQ的密码和虚拟主机
spring.rabbitmq.addresses=120.79.18.77 spring.rabbitmq.port=5672 spring.rabbitmq.virtual-host=/ -
启用RabbitMQ
@SpringBootApplication @EnableDiscoveryClient @EnableFeignClients @EnableRabbit public class GrainmallWareApplication { public static void main(String[] args) { SpringApplication.run(GrainmallWareApplication.class, args); } } -
配置对象的JSON序列化编码
@Configuration public class RabbitConfig { @Autowired RabbitTemplate rabbitTemplate; @Bean public MessageConverter messageConverter() { return new Jackson2JsonMessageConverter(); } } -
创建交换机、路由键和消息队列：
@Bean public Exchange stockEventExchange() { //String name, boolean durable, boolean autoDelete, Map<String, Object> arguments return new TopicExchange("stock-event-exchange", true, false); } @Bean public Queue stockDelayQueue() { //String name, boolean durable, boolean exclusive, boolean autoDelete, @Nullable Map<String, Object> arguments Map<String, Object> args = new HashMap<>(); args.put("x-dead-letter-exchange", "stock-event-exchange"); args.put("x-dead-letter-routing-key", "stock.release"); args.put("x-message-ttl", 120000); return new Queue("stock.delay.queue", true, false, false, args); } @Bean public Queue stockReleaseQueue() { //String name, boolean durable, boolean exclusive, boolean autoDelete, @Nullable Map<String, Object> arguments return new Queue("stock.release.stock.queue", true, false, false); } @Bean public Binding stockReleaseBinding() { //String destination, Binding.DestinationType destinationType, String exchange, String routingKey, @Nullable Map<String, Object> arguments return new Binding("stock.release.stock.queue", Binding.DestinationType.QUEUE, "stock-event-exchange", "stock.release.#", null); } @Bean public Binding stockLockedBinding() { //String destination, Binding.DestinationType destinationType, String exchange, String routingKey, @Nullable Map<String, Object> arguments return new Binding("stock.delay.queue", Binding.DestinationType.QUEUE, "stock-event-exchange", "stock.locked", null); } -
创建监听器（否则不会创建以上的RabbitMQ组件）
@RabbitListener(queues = "stock.release.stock.queue") public void listener(Message message) { } -
延时队列实现完成：
4.3 监听库存解锁
库存解锁的场景：
- 下订单成功后，订单过期没有被支付从而被系统自动取消、被用户手动取消，都需要解锁库存
- 下订单成功后，库存锁定成功，但接下来业务调用失败导致订单回滚，那么之前锁定的库存都要自动回滚
4.4 订单回滚情况下的库存解锁
步骤：
-
创建
工作详情单：库存锁定成功后，首先要保存一份库存工作详情单单，库存工作单的目的是为了回溯库存锁定情况，即记录哪个商品、在哪一个库存锁定了多少数量
@Data @NoArgsConstructor @AllArgsConstructor @TableName("wms_ware_order_task_detail") public class WmsWareOrderTaskDetailEntity implements Serializable { private static final long serialVersionUID = 1L; /** * id */ @TableId private Long id; /** * sku_id */ private Long skuId; /** * sku_name */ private String skuName; /** * 购买个数 */ private Integer skuNum; /** * 工作单id */ private Long taskId; /** * 仓库id */ private Long wareId; /** * 1-已锁定 2-已解锁 3-扣减 */ private Integer lockStatus; } -
创建
工作单：一个订单对应一个工作单，且一个工作单对应多个工作详情单（类似订单和订单项的关系）@Data @TableName("wms_ware_order_task") public class WmsWareOrderTaskEntity implements Serializable { private static final long serialVersionUID = 1L; /** * id */ @TableId private Long id; /** * order_id */ private Long orderId; /** * order_sn */ private String orderSn; /** * 收货人 */ private String consignee; /** * 收货人电话 */ private String consigneeTel; /** * 配送地址 */ private String deliveryAddress; /** * 订单备注 */ private String orderComment; /** * 付款方式【 1:在线付款 2:货到付款】 */ private Integer paymentWay; /** * 任务状态 */ private Integer taskStatus; /** * 订单描述 */ private String orderBody; /** * 物流单号 */ private String trackingNo; /** * create_time */ private Date createTime; /** * 仓库id */ private Long wareId; /** * 工作单备注 */ private String taskComment; } -
如果订单中的每一件商品都锁定库存成功，就将当前商品锁定了几件商品、在哪个库存锁定的工作详情单记录发送给MQ
-
因此定义一个如下的数据结构
StockLockedTO，包括库存工作单的id和库存工作详情单id，每当锁定成功一个商品的库存，就向MQ发送一个消息：@Data public class StockLockedTO { // 库存工作单id private Long taskId; // 库存工作单详情id private Long taskDetailId; } 下面的代码的意思是，遍历所有有本件商品库存的仓库（wareIds），如果锁库存（wareSkuDao.lockSkuStock）成功，则保存
库存工作详情单（设置其taskId为工作单id、保存这次锁库存锁的skuId、锁了多少件、sku的name、在哪锁的：仓库id以及状态 1-已锁定 2-已解锁 3-扣减）for(Long wareId : wareIds) { Long res = wareSkuDao.lockSkuStock(skuId, wareId, num); if(res == 1) { hasStockFlag = true; WmsWareOrderTaskDetailEntity orderTaskDetailEntity = new WmsWareOrderTaskDetailEntity(null, skuId, hasStock.getSkuName(), hasStock.getNum(), orderTaskEntity.getOrderId(), wareId, 1); // 1表示锁定成功 // 保存库存工作详情单 orderTaskDetailService.save(orderTaskDetailEntity); StockLockedTO stockLockedTO = new StockLockedTO(); // 设置工作单id stockLockedTO.setTaskId(orderTaskEntity.getId()); // 设置工作详情单的id stockLockedTO.setTaskDetailId(orderTaskDetailEntity.getId()); // 通知MQ库存锁定成功 rabbitTemplate.convertAndSend("stock-event-exchange", "stock.locked", stockLockedTO ); break; } // 仓库锁定失败，尝试下一个仓库 } -
如果有一件锁定失败，则之前保存的
工作详情单就回滚了，但是却把这些工作详情单的消息发送出去了。其实这个是没有影响的，因为等待消息成为死信后，会重新拿着工作详情单的id去工作详情单查询，由于回滚是怎么也查不到的，因此不会有任何业务逻辑的影响，但是会多少增加数据库的压力
更好的解决办法是不仅仅保存工作详情单的id，而是保存整个工作详情单，这样做的好处有：
- 不需要再拿着工作详情单的id去工作详情单查询
- 如果此时库存锁定失败，且库存工作详情单和库存不在一个数据库，则工作详情单回滚但是库存表并不会回滚，导致的结果是无法利用工作详情单去解锁库存了
-
因此将
StockLockedTO修改为：@Data public class StockLockedTO { // 库存工作单id private Long taskId; // 库存工作详情单 private StockDetailTO stockDetailTO; }其中的
StockDetailTO是抽取ware服务中的WmsWareOrderTaskDetailEntity到common下的：public class StockDetailTO { /** * id */ @TableId private Long id; /** * sku_id */ private Long skuId; /** * sku_name */ private String skuName; /** * 购买个数 */ private Integer skuNum; /** * 工作单id */ private Long taskId; /** * 仓库id */ private Long wareId; /** * 1-已锁定 2-已解锁 3-扣减 */ private Integer lockStatus; }这样代码就修改成：“发送包括整个库存详情单而不是详情单id的消息”
for(Long wareId : wareIds) { Long res = wareSkuDao.lockSkuStock(skuId, wareId, num); if(res == 1) { hasStockFlag = true; WmsWareOrderTaskDetailEntity orderTaskDetailEntity = new WmsWareOrderTaskDetailEntity(null, skuId, hasStock.getSkuName(), hasStock.getNum(), orderTaskEntity.getOrderId(), wareId, 1); // 1表示锁定成功 // 保存库存工作详情单 orderTaskDetailService.save(orderTaskDetailEntity); StockLockedTO stockLockedTO = new StockLockedTO(); // 设置工作单id stockLockedTO.setTaskId(orderTaskEntity.getId()); // 设置工作详情单的id StockDetailTO stockDetailTO = new StockDetailTO(); BeanUtils.copyProperties(orderTaskDetailEntity, stockDetailTO); stockLockedTO.setStockDetailTO(stockDetailTO); // 通知MQ库存锁定成功 rabbitTemplate.convertAndSend("stock-event-exchange", "stock.locked", stockLockedTO ); break; } // 仓库锁定失败，尝试下一个仓库 } -
测试：回想一下之前的代码，我在订单服务成功锁定库存后执行了一步除零操作模拟后续业务异常，并且去掉了分布式事务的注解，因此正常情况下此时订单会回滚，但是库存不会回滚（远程调用），这样就可以利用库存执行向消息队列发送的消息（其中的库存工作详情单）来回滚锁定的库存
首先提交订单前端异常：
检查订单并没有创建（订单回滚了）：
但是库存却被锁定了（远程事务没有回滚）：
此时stock的延时队列中存在着库存锁定成功发来的消息：
-
接下来就是通过这个变成死信的消息来自动解锁库存，实现订单创建失败情况下的库存最终一致性
4.5 库存自动解锁
-
首先创建一个
Listener，用于监听库存释放队列stock.release.stock.queue，如果解锁成功，需要手动ack，这里的两个参数为：deliveryTag：存在于接收方收到的消息的消息头中，表示消息的唯一标识multiple：是否开启批量签收
@Service @RabbitListener(queues = "stock.release.stock.queue") @Slf4j public class StockReleaseListener { @Autowired WmsWareSkuService wareSkuService; @RabbitHandler public void handleStockLockedRelease(StockLockedTO to, Message message, Channel channel) throws IOException { log.info("收到库存解锁消息：" + message); try { wareSkuService.unlockStock(to, message, channel); channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); log.info("库存自动解锁成功：" + message); } catch (IOException e) { e.printStackTrace(); log.info("库存自动解锁异常，消息重新放入队尾：" + message); // 出现异常就放回队尾 channel.basicReject(message.getMessageProperties().getDeliveryTag(), true); } } } -
如果过程中出现异常，则拒签消息，
basicReject和basicUAck的区别在于后者支持批量，第二个参数表示拒绝签收消息后是否放入队列 -
具体的库存解锁业务如下：
- 首先根据消息中的
库存工作单数据，去查找工作详情单记录数据和工作单对应的订单记录数据 - 如果数据库中不存在该
工作详情单记录记录或者工作详情单记录不为锁定的状态，则表明工作详情单无效直接丢弃消息即可 - 否则查看订单的状态，如果订单的状态为已关闭或者异常状态，表明订单数据在提交的时候进行了回滚但是库存数据没有，这样就利用
工作详情单记录回滚库存数据，保证库存数据的最终一致性 - 处理过程出现异常的情况也会被tay catch到然后将消息重新放入队尾
/** * 根据工作详情单解锁库存 * @param to * @param message * @param channel * @throws IOException */ @Override public void unlockStock(StockLockedTO to, Message message, Channel channel) throws IOException { Long taskId = to.getTaskId(); StockDetailTO stockDetailTO = to.getStockDetailTO(); // 解锁 // 查询数据库的库存工作详情单种关于这个订单的锁定库存信息 WmsWareOrderTaskDetailEntity taskDetailEntity = orderTaskDetailService.getById(stockDetailTO.getId()); if(taskDetailEntity != null && taskDetailEntity.getLockStatus() == 1) { // 解锁库存 WmsWareOrderTaskEntity taskEntity = orderTaskService.getById(taskId); String orderSn = taskEntity.getOrderSn(); // 解不解锁要看查看订单的状态 R r = orderFeignService.reqOrderByOrderSn(orderSn); if(r.getCode() == 200) { OrderEntityTO orderEntity = r.getData("data", new TypeReference<OrderEntityTO>() { }); if(orderEntity == null) { unlockStock(taskDetailEntity); } else { // 订单状态【0->待付款；1->待发货；2->已发货；3->已完成；4->已关闭；5->无效订单】 Integer status = orderEntity.getStatus(); // 订单关闭或者无效，则解锁库存 if(status == 4 || status == 5) { unlockStock(taskDetailEntity); } } } else { throw new RuntimeException("订单查询服务调用失败"); } } } /** * 解锁库存 并 修改工作单状态为已解锁 * @param taskDetailEntity */ private void unlockStock(WmsWareOrderTaskDetailEntity taskDetailEntity) { // 解锁库存 wareSkuDao.unlockStockByTaskDetail(taskDetailEntity); // 改变工作详情单的状态 WmsWareOrderTaskDetailEntity taskDetailEntity1 = new WmsWareOrderTaskDetailEntity(); taskDetailEntity1.setId(taskDetailEntity.getId()); taskDetailEntity1.setLockStatus(2); orderTaskDetailService.updateById(taskDetailEntity1); } - 首先根据消息中的
4.6 踩的坑
-
两个消费者导致队列消息一直处于
unacked的状态，即如果有两个listener同时在监听一个队列，那么队列将无法工作。队列的消费者数量可以通过网页查看：
-
消费端在锁定库存的时候，远程查询订单状态会被拦截器鉴权拦截：这不单单是feign的问题，因为这个远程调用属于服务端发起的，本身也不带有身份token，因此需要再orderService中的拦截器配置路径放行：
@Override public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception { String token = request.getHeader("token"); CheckResult checkResult = JWTUtil.validateJwt(token); String requestURI = request.getRequestURI(); boolean match = new AntPathMatcher().match("/order/order/status/**", requestURI); if(match) { return true; } ... }
5 定时关单完成
- 在创建订单完成后，会进行库存的锁定，然后订单系统向MQ发送消息以进行自动关单，库存系统在锁定库存后也会向MQ发送消息以进行自动解锁库存。
- 一般情况下，订单延时队列的TTL要比库存延时队列的TTL要断，因此正常是先关闭订单，库存到时间后判断订单状态进行关闭。
- 但是如果订单系统由于网络异常、延迟的情况导致消息到达MQ的时间比库存系统的消息完，就会导致库存查看订单状态时仍为新建状态而不进行任何处理，导致库存一直得不到解锁
- 解决方法就是在订单过期，订单关闭的时候再发送一次消息给MQ，库存监听到消息后就去解锁库存
@RabbitListener(queues = "order.release.order.queue")
@Service
@Slf4j
public class OrderCloseListener {
@Autowired
OrderService orderService;
@RabbitHandler
public void orderCloseHandle(OrderEntity orderEntity, Channel channel, Message message) throws IOException {
log.info("收到过期订单消息，准备关闭订单：" + orderEntity);
try {
orderService.closeOrder(orderEntity);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
log.error(e.getMessage());
channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
}
}
}
/**
* 关闭订单
* @param orderEntity
*/
@Override
public void closeOrder(OrderEntity orderEntity) {
// 获取数据库最新的订单状态
OrderEntity order = this.getById(orderEntity.getId());
if(OrderStatusEnum.CREATE_NEW.getCode().equals(order.getStatus())) {
// 只更新id和status字段
OrderEntity newOrderEntity = new OrderEntity();
newOrderEntity.setId(order.getId());
// 更新状态为已取消
newOrderEntity.setStatus(OrderStatusEnum.CANCLED.getCode());
this.updateById(newOrderEntity);
OrderEntityTO orderEntityTO = new OrderEntityTO();
BeanUtils.copyProperties(order, orderEntityTO);
// 关单后再向MQ发送一个消息，用于解锁库存
rabbitTemplate.convertAndSend("order-event-exchange",
"order.release.other.#",
orderEntityTO);
}
}
仓库服务监听队列，根据库存工作详情单解锁库存
/**
* 订单关闭，自动解锁库存
* 防止订单服务卡顿，导致订单状态无法修改（始终处于新建状态）
* 库存消息达到队列后，查到订单为新建状态就不做任何处理
* 从而导致库存永远无法解锁
* @param orderEntity
* @param message
* @param channel
*/
@Override
public void unlockStock(OrderEntityTO orderEntity, Message message, Channel channel) {
String orderSn = orderEntity.getOrderSn();
// 走到这里就说明订单创建成功了
// 查一下库存解锁状态，防止重复解锁库存
WmsWareOrderTaskEntity orderTaskBySn
= orderTaskService.getOrderTaskBySn(orderSn);
List<WmsWareOrderTaskDetailEntity> list = orderTaskDetailService.list(new LambdaQueryWrapper<WmsWareOrderTaskDetailEntity>()
.eq(WmsWareOrderTaskDetailEntity::getTaskId, orderTaskBySn.getOrderSn())
.eq(WmsWareOrderTaskDetailEntity::getLockStatus, 1));
list.forEach(this::unlockStock);
}
6 保证消息的可靠性：消息丢失、积压、重复问题的解决
前面是使用RabbitMQ可靠消息+最终一致性解决方案的柔性事务的解决方案来替代Seata解决分布式事务的，因此保持RabbitMQ的消息的可靠性就至关重要：
6.1 消息丢失
-
消息发送出去，但是由于网络原因没有抵达消息队列服务器
-
做好容错方案，如失败采用重试机制，在消息发送之前记录到数据库，然后利用确认机制在消息到达Broker后修改记录的消息状态，采用定期扫描重发未到达ed消息的方式解决这个问题
-
做好日志记录，即每个消息状态是否被服务器收到都应该被记录
-
做好定期重发，如果消息没有发送成功，定期去数据库扫描未成功的消息进行重发
-
-
消息抵达Broker，但是Broker还没来得及将消息写入磁盘进行持久化就宕机了
- 发布者加入确认回调机制，确认成功的消息然后修改数据库状态，同样针对未持久化的数据定期扫描数据库进行重发
-
自动ACK状态下，消费者收到消息，但还没来得及回消息就宕机了，导致自动ACK消息丢失
- 开启手动ACK，只有消费成功才移除消息，失败或者没来得及消费则将消息NOAck重新加入队尾
6.2 消息重复
- 消息消费成功，事务已经提交但是在ack的时候，机器宕机导致ack失败。然后Broker的消息由unack变为ready，并发送给其他消费者（执行了两次）
- 消费者的接口应该设置为幂等性的，比如扣库存的时候是先判断库存工作单的工作状态，因此不可能执行两次
- 使用防重表（redis/mysql），发送的每一个消息都拥有业务的唯一标识，处理过后就不再处理了
- rabbitMQ的每一个消息都有
redelivered字段，可以用来判断是是否是被重新投递过来的
- 消息消费失败，由于重试机制，自动又将消息发送出去（执行了一次，这种情况是正确的）
6.3 消息积压
- 消费者宕机导致的积压
- 消费者消费能力不足导致的积压
- 发送者流量太大导致的积压
- 上线更多消费者，进行正常消费
- 上线专门的队列消费服务，先将消息批量抽取出来，记录数据库，再慢慢离线处理
- 对流量进行限流处理