尚品汇-延迟插件实现订单超时取消（四十五）

目录：

（1）延迟插件封装

（2）基于延迟插件测试

        如何保证消息幂等性？

（3）改造订单service-order模块-实现订单超时取消

（1）延迟插件封装

        

把消息带过去： 

在消息的重试发送消息的方法里封装：retrySendMsg

（2）基于延迟插件测试

service-order模块

 rabbit-util模块配置常量MqConst

/*** 取消订单，发送延迟队列*/
public static final String EXCHANGE_DIRECT_ORDER_CANCEL = "exchange.direct.order.cancel";//"exchange.direct.order.create" test_exchange;
public static final String ROUTING_ORDER_CANCEL = "order.create";
//延迟取消订单队列
public static final String QUEUE_ORDER_CANCEL  = "queue.order.cancel";
//取消订单 延迟时间 单位：秒
public static final int DELAY_TIME  = 10;

rabbit-util模块延迟接口封装：RabbitService

/*** 封装发送延迟消息方法* @param exchange* @param routingKey* @param msg* @param delayTime* @return*/
public Boolean sendDelayMsg(String exchange,String routingKey, Object msg, int delayTime){//  将发送的消息 赋值到 自定义的实体类GmallCorrelationData gmallCorrelationData = new GmallCorrelationData();//  声明一个correlationId的变量String correlationId = UUID.randomUUID().toString().replaceAll("-","");gmallCorrelationData.setId(correlationId);gmallCorrelationData.setExchange(exchange);gmallCorrelationData.setRoutingKey(routingKey);gmallCorrelationData.setMessage(msg);gmallCorrelationData.setDelayTime(delayTime);gmallCorrelationData.setDelay(true);//  将数据存到缓存this.redisTemplate.opsForValue().set(correlationId,JSON.toJSONString(gmallCorrelationData),10,TimeUnit.MINUTES);//  发送消息this.rabbitTemplate.convertAndSend(exchange,routingKey,msg,message -> {//  设置延迟时间message.getMessageProperties().setDelay(delayTime*1000);return message;},gmallCorrelationData);//  默认返回return true;
}

修改retrySendMsg方法 – 添加判断是否属于延迟消息

 MQProducerAckConfig 配置类中修改retrySendMsg方法

//  判断是否属于延迟消息
if (gmallCorrelationData.isDelay()){//  属于延迟消息this.rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(),gmallCorrelationData.getRoutingKey(),gmallCorrelationData.getMessage(),message -> {//  设置延迟时间message.getMessageProperties().setDelay(gmallCorrelationData.getDelayTime()*1000);return message;},gmallCorrelationData);
}else {//  调用发送消息方法 表示发送普通消息  发送消息的时候，不能调用 new RabbitService().sendMsg() 这个方法this.rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(),gmallCorrelationData.getRoutingKey(),gmallCorrelationData.getMessage(),gmallCorrelationData);
}

 Contrroller：

利用封装好的工具类 测试发送延迟消息

//  基于延迟插件的延迟消息
@GetMapping("sendDelay")
public Result sendDelay(){//  声明一个时间对象SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("发送时间："+simpleDateFormat.format(new Date()));this.rabbitService.sendDelayMsg(DelayedMqConfig.exchange_delay,DelayedMqConfig.routing_delay,"iuok",3);return Result.ok();
}

消息没有成功到达队列，会出发回调重新发送，会尝试发送三次，消费者会消费三次 

结果会 回发送三次，也被消费三次！

如何保证消息幂等性？

1. 使用数据方式
2. 使用redis setnx 命令解决  --- 推荐

幂等性：执行多次，结果都是一样的

在消费者这里进行实现，消费者不管发送者发送多少条消息，只消费一次

 

@SneakyThrows
@RabbitListener(queues = DelayedMqConfig.queue_delay_1)
public void getMsg2(String msg,Message message,Channel channel){//  使用setnx 命令来解决 msgKey = delay:iuokString msgKey = "delay:"+msg;Boolean result = this.redisTemplate.opsForValue().setIfAbsent(msgKey, "0", 10, TimeUnit.MINUTES);//  result = true : 说明执行成功，redis 里面没有这个key ，第一次创建， 第一次消费。//  result = false : 说明执行失败，redis 里面有这个key//  能： 保证消息被消费成功    第二次消费，可以进来，但是要判断上一个消费者，是否将消息消费了。如果消费了，则直接返回，如果没有消费成功，我消费。//  在设置key 的时候给了一个默认值 0 ，如果消费成功，则将key的值 改为1if (!result){//  获取缓存key对应的数据String status = (String) this.redisTemplate.opsForValue().get(msgKey);if ("1".equals(status)){//  手动确认channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);return;} else {//  说明第一个消费者没有消费成功，所以消费并确认SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("接收时间："+simpleDateFormat.format(new Date()));System.out.println("接收的消息："+msg);//  修改redis 中的数据this.redisTemplate.opsForValue().set(msgKey,"1");channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);return;}}SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("接收时间："+simpleDateFormat.format(new Date()));System.out.println("接收的消息："+msg);//  修改redis 中的数据this.redisTemplate.opsForValue().set(msgKey,"1");//  手动确认消息channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}

发送一次消息 

在发送一次消息：Redis有了就不会消费了

（3）改造订单service-order模块-实现订单超时取消

 ​​​​​​​​​​​​​

 

service-order模块配置队列

添加依赖

<!--rabbitmq消息队列-->
<dependency><groupId>com.atguigu.gmall</groupId><artifactId>rabbit-util</artifactId><version>1.0</version>
</dependency>

 

OrderCanelMqConfig  

package com.atguigu.gmall.order.receiver;@Configuration
public class OrderCanelMqConfig {@Beanpublic Queue delayQueue() {// 第一个参数是创建的queue的名字，第二个参数是是否支持持久化return new Queue(MqConst.QUEUE_ORDER_CANCEL, true);}@Beanpublic CustomExchange delayExchange() {Map<String, Object> args = new HashMap<String, Object>();args.put("x-delayed-type", "direct");return new CustomExchange(MqConst.EXCHANGE_DIRECT_ORDER_CANCEL, "x-delayed-message", true, false, args);}@Beanpublic Binding bindingDelay() {return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(MqConst.ROUTING_ORDER_CANCEL).noargs();}
}

发送消息

创建订单时，发送延迟消息

OrderServiceImpl实现类中：

修改保存订单方法

 

@Override
@Transactional
public Long saveOrderInfo(OrderInfo orderInfo) {.....//发送延迟队列，如果定时未支付，取消订单//交换机、路由key、消息（订单id） 超时时间rabbitService.sendDelayMessage(MqConst.EXCHANGE_DIRECT_ORDER_CANCEL, MqConst.ROUTING_ORDER_CANCEL, orderInfo.getId(), MqConst.DELAY_TIME);// 返回return orderInfo.getId();
}

 

接收消息

传的是订单id，这里orderId会接受到，赋值给他，message不是接受到的消息

package com.atguigu.gmall.order.receiver;@Component
public class OrderReceiver {@Autowiredprivate OrderService orderService;//  监听的消息@SneakyThrows@RabbitListener(queues = MqConst.QUEUE_ORDER_CANCEL)public void cancelOrder(Long orderId , Message message, Channel channel){//  判断当前订单Id 不能为空try {if (orderId!=null){//  发过来的是订单Id，那么你就需要判断一下当前的订单是否已经支付了。//  未支付的情况下：关闭订单//  根据订单Id 查询orderInfo select * from order_info where id = orderId//  利用这个接口IService  实现类ServiceImpl 完成根据订单Id 查询订单信息 ServiceImpl 类底层还是使用的mapperOrderInfo orderInfo = orderService.getById(orderId);//  判断支付状态,进度状态if (orderInfo!=null && "UNPAID".equals(orderInfo.getOrderStatus())&& "UNPAID".equals(orderInfo.getProcessStatus())){//  关闭订单//  int i = 1/0;orderService.execExpiredOrder(orderId);}}} catch (Exception e) {//  消息没有正常被消费者处理： 记录日志后续跟踪处理! e.printStackTrace();}//  手动确认消息 如果不确认，有可能会到消息残留。channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}
}

package com.atguigu.gmall.model.enums;public enum ProcessStatus {UNPAID("未支付", OrderStatus.UNPAID),PAID("已支付", OrderStatus.PAID),NOTIFIED_WARE("已通知仓储", OrderStatus.PAID),WAITING_DELEVER("待发货", OrderStatus.WAITING_DELEVER),STOCK_EXCEPTION("库存异常", OrderStatus.PAID),DELEVERED("已发货", OrderStatus.DELEVERED),CLOSED("已关闭", OrderStatus.CLOSED),COMMNET("已评价",OrderStatus.FINISHED) ,FINISHED("已完结", OrderStatus.FINISHED) ,PAY_FAIL("支付失败", OrderStatus.UNPAID),SPLIT("订单已拆分", OrderStatus.SPLIT);private String comment ;private OrderStatus orderStatus;ProcessStatus(String comment, OrderStatus orderStatus){this.comment=comment;this.orderStatus=orderStatus;}public String getComment() {return comment;}public void setComment(String comment) {this.comment = comment;}public OrderStatus getOrderStatus() {return orderStatus;}public void setOrderStatus(OrderStatus orderStatus) {this.orderStatus = orderStatus;}}

 

编写取消订单接口与实现类

/*** 处理过期订单* @param orderId*/
void execExpiredOrder(Long orderId);
/*** 根据订单Id 修改订单的状态* @param orderId* @param processStatus*/
void updateOrderStatus(Long orderId, ProcessStatus processStatus);

 

@Override
public void execExpiredOrder(Long orderId) {// orderInfoupdateOrderStatus(orderId, ProcessStatus.CLOSED);
}
@Override
public void updateOrderStatus(Long orderId, ProcessStatus processStatus) {OrderInfo orderInfo = new OrderInfo();orderInfo.setId(orderId);orderInfo.setProcessStatus(processStatus.name());orderInfo.setOrderStatus(processStatus.getOrderStatus().name());orderInfoMapper.updateById(orderInfo);
}

 

 

最终会变成Close