消息队列处理微信支付超时订单

╰半夏微凉° 2021-09-23 05:04 319阅读 0赞

**1.配置交换机、队列  **

RabbitMqConfig

/**
     *      * 延时队列交换机
     *      * 注意这里的交换机类型：CustomExchange
     *      *
     *      * @return
     *      
     */
    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        //属性参数 交换机名称 交换机类型 是否持久化 是否自动删除 配置参数
        return new CustomExchange("delay_exchange", "x-delayed-message", true, false, args);
    }

/**
     * * 延时队列
     * * @return    
     */
    @Bean
    public Queue delayOrderQueue() {
        //属性参数 队列名称 是否持久化
        return new Queue("delay_order", true);
    }
    
    /**
     *      * 给延时队列绑定交换机
     *      *
     *      * @return
     *      
     */
    @Bean
    public Binding orderDelayBinding() {
        return BindingBuilder.bind(delayOrderQueue()).to(delayExchange()).with("order").noargs();
    }

**2.支付接口调用，创建订单的时候设置订单过期时间**

public Object call() throws Exception {
        this.validate();
        if (apiResult.getCode() != 0) {
            return apiResult;
        }
    
        Map<String, Object> map = this.unifiedOrder();
        if (map == null) {
            apiResult.fail("支付请求失败！");
            logger.info("预支付请求失败");
            return apiResult;
        }
        if (!"SUCCESS".equals(map.get("return_code"))) {
            apiResult.fail((String) map.get("return_msg"));
            return apiResult;
        }
        if (!(map.get("sign")).equals(SignUtil.getSign(map, payConfig.getKey()))) {
            apiResult.fail("支付请求失败！");
            logger.info("预支付请求失败：{}", "验签未通过");
            return apiResult;
        }
        if (!"SUCCESS".equals(map.get("result_code"))) {
            apiResult.fail((String) map.get("err_code_des"));
            logger.info("预支付请求失败：{}", map.get("err_code_des"));
            return apiResult;
        }
    
    
        //创建订单
        this.createOrder(map);
    
    
        //返回预支付参数
        String prepay_id = (String) map.get("prepay_id");
        Map<String, Object> result = new HashMap<>();
        result.put("appId", payConfig.getAppid());
        result.put("nonceStr", NumberUtil.getRandomStr(16));
        result.put("package", "prepay_id=" + prepay_id);
        result.put("signType", "MD5");
        result.put("timeStamp", System.currentTimeMillis() / 1000);
        result.put("sign", SignUtil.getSign(result, payConfig.getKey()));
        logger.info("预支付请求接口返回参数：{}", JsonUtil.toJson(result));
        result.put("orderId", orderId);
        //第一个参数是前面RabbitMqConfig的交换机名称 第二个参数的路由名称 第三个参数是传递的参数 第四个参数是配置属性
        template.convertAndSend("delay_exchange", "order", orderNo, message -> {
            //配置消息的过期时间
            message.getMessageProperties().setDelay(35 * 60 * 100);
            return message;
        });
    
        apiResult.succ(result);
        return apiResult;
    }

**3.创建监听队列，监听指定类型的消息**

//用户超过三分钟未支付删除订单
    @RabbitListener(queues = "delay_order")
    public void consumeOrderMessage(String orderNo, @Header(AmqpHeaders.DELIVERY_TAG) long tag, Channel channel) {
        logger.info("3分钟延迟队列处理订单：{}", orderNo);
        try {
            ActivityOrder order = activityOrderService.queryByOrderNo(orderNo);
            if (order == null) {
                channel.basicAck(tag, true);
                return;
            }
            if (ActivityPayEnum.WAIT_PAYING.getValue().equals(order.getOrderStatus())) {
                order.setValid(false);
                order.setUpdateTime(new Date());
                activityOrderService.updateActivityOrder(order);
                channel.basicAck(tag, true);
            } else {
                logger.info("订单状态已经流转：{}", orderNo);
                channel.basicAck(tag, true);
            }
        } catch (Exception e) {
            logger.error("消费订单延迟队列消息时抛出异常 ， e = {}", e.getMessage());
            //重新放回到队列里面
            try {
                Thread.sleep(50);
    
                channel.basicNack(tag, false, true);
            } catch (IOException | InterruptedException e1) {
                logger.error("消费订单延迟队列消息时抛出异常 ， e = {}" + e.getMessage());
            }
            e.printStackTrace();
        }
    
    }