网站 实名认证腾讯云域名申请
目录：
（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();
}消息没有成功到达队列，会出发回调重新发送，会尝试发送三次，消费者会消费三次
结果会 回发送三次，也被消费三次！
如何保证消息幂等性？
- 使用数据方式
- 使用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