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

目录：

（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:iuok
    String 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的值 改为1
    if (!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 {

    @Bean
    public Queue delayQueue() {
        // 第一个参数是创建的queue的名字，第二个参数是是否支持持久化
        return new Queue(MqConst.QUEUE_ORDER_CANCEL, true);
    }

    @Bean
    public 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);
    }

    @Bean
    public 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 {

    @Autowired
    private 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 类底层还是使用的mapper
                OrderInfo 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) {
    // orderInfo
    updateOrderStatus(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);
}