消息延迟队列处理拼团时间到期

深碍√TFBOYSˉ_ 2021-09-24 19:36 304阅读 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 delayTuanQueue(){
        //属性参数 队列名称 是否持久化
        return new Queue("delay_tuan", true);
    }
    
    /**
     * 给延时队列绑定交换机
     * @return
     */
    @Bean
    public Binding tuanDelayBinding(){
        return BindingBuilder.bind(delayTuanQueue()).to(delayExchange()).with("tuan").noargs();
    }

2.在拼团活动订单支付回调接口里面触发，支付回调接口

@Transactional
    public String callback(HttpServletRequest request) {
    
        String result = "";
        Date date = new Date();
        try (Jedis jedis = jedisPool.getResource();
             InputStream inStream = request.getInputStream();
             ByteArrayOutputStream outSteam = new ByteArrayOutputStream()) {
            byte[] buffer = new byte[1024];
            int len = 0;
            while ((len = inStream.read(buffer)) != -1) {
                outSteam.write(buffer, 0, len);
            }
            result = new String(outSteam.toByteArray(), StandardCharsets.UTF_8);
            logger.info("wechat callback result:" + result);
            //处理返回结果
            Map<String, Object> data = XmlUtil.xmlToMap(result);
            String orderNo = (String) data.get("out_trade_no");
            String transactionId = (String) data.get("transaction_id");
            logger.info("回调订单号：{}", orderNo);
            ActivityOrder order = activityOrderService.queryByOrderNo(orderNo);
            if (order == null) {
                return failReturn;
            }
            PayConfig payConfig = payConfigService.queryByToken(order.getProgramToken());
            if (!SignUtil.getSign(data, payConfig.getKey()).equals(data.get("sign"))) {
                logger.info("处理返回结果失败:验签错误");
                return failReturn;
            }
            logger.info("回调参数：{}", JsonUtil.toJson(data));
    
            if (order.getStatus().equals(PayEnum.PAYING_SUSS.getValue())) {
                return successReturn;
            }
            order.setCallBackStr(result);
            order.setWechatNo(transactionId);
            logger.info("回调参数保存成功");
    
            //处理返回数据
            if (!"SUCCESS".equals(data.get("return_code"))) {
                logger.info("支付失败");
                order.setOrderStatus(PayEnum.PAYING_FAIL.getValue());
                order.setUpdateTime(new Date());
                activityOrderService.updateActivityOrder(order);
                return successReturn;
            }
            String checkNo = NumberUtil.getRandomNumber(8);
            while (true) {
                if (activityOrderService.countByCheckNo(checkNo, order.getShopId()) > 0) {
                    checkNo = NumberUtil.getRandomNumber(8);
                } else {
                    break;
                }
            }
            order.setCheckNo(checkNo);
            order.setOrderStatus(PayEnum.PAYING_SUSS.getValue());
            order.setUpdateTime(new Date());
            order.setPayTime(new Date());
            MiniProgram miniProgram = miniProgramService.queryByToken(order.getProgramToken());
            UserWechatAccessToken userWechatAccessToken = userWechatAccessTokenService.queryByAppid(miniProgram.getAppid());
    
            String token = userWechatAccessToken == null ? "" : userWechatAccessToken.getAccessToken();
    
            Activity activity = activityService.getActivity(order.getActivityId());
    
            if(order.getIsMain() && activity.getTimeLimit()){
                //设置延迟时间
                long minute=activity.getDayLimit()*24*60+activity.getHourLimit()*60+activity.getMinuteLimit();
                //第一个参数是前面RabbitMqConfig的交换机名称 第二个参数的路由名称 第三个参数是传递的参数 第四个参数是配置属性
                template.convertAndSend("delay_exchange", "tuan", order.getGroupNo(), message -> {
                    //配置消息的过期时间
                    message.getMessageProperties().setDelay((int) (minute * 60 * 1000));
                    return message;
                });
            }
    
            String path = "";
            if (ActivityTypeEnum.GROUP.getName().equals(activity.getaType())) {
                path = "pages/joinActive3/main";
            } else if (ActivityTypeEnum.INVITE_CARD.getName().equals(activity.getaType())) {
                path = "pages/joinZan1/main";
            }
            order.setQrCode(imageBo.getQrCode(order.getId(), order.getActivityId(), path, token));
            activityOrderService.updateActivityOrder(order);
            logger.info("更新订单结果");
            return successReturn;
        } catch (Exception e) {
            logger.info("sbsbsbsbsbs:{}", failReturn);
            e.printStackTrace();
            logger.info(e.getMessage());
            return failReturn;
        }
    
    }

3.OrderConsumer

//拼团时间到期处理未成团
    @RabbitListener(queues = "delay_tuan")
    public void consumeTuanOrderMessage(String groupNo, @Header(AmqpHeaders.DELIVERY_TAG) long tag, Channel channel) {
        logger.info("拼团时间到期处理未成团活动订单：{}", groupNo);
        try {
            List<ActivityOrder> orderList=activityOrderService.queryByGroupNo(groupNo);
            if (orderList == null || orderList.size()==0) {
                channel.basicAck(tag, true);
                return;
            }
            Activity activity=activityService.getActivity(orderList.get(0).getActivityId());
            PayConfig payConfig=payConfigService.queryByToken(activity.getProgramToken);
            if(!activity.getAutoGroup()){
                if(orderList.size()<activity.getMinAmount()){
                    for(ActivityOrder order:orderList){
                        logger.info("开始处理拼团到期退款：拼团号{}"，groupNo);
                        refund(order.getOrderNo(), payConfig);
                        
                    }
                }
            }
            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();
        }
    
    }

退款

public void refund(String orderNo, PayConfig payConfig){
        ActivityOrder order = activityOrderService.queryByOrderNo(orderNo);
        String refundNo = OrderNoUtil.generateRefundOrderNo();
        //准备预支付
        Map<String, Object> param = new HashMap<>();
        param.put("appid", payConfig.getAppid());
        param.put("mch_id", payConfig.getMchId());
        param.put("nonce_str", NumberUtil.getRandomStr(18));
        param.put("out_trade_no", orderNo);
        param.put("total_fee", order.getOrderAmount());
        param.put("refund_fee", order.getOrderAmount());
        param.put("out_refund_no", refundNo);
        String sign = SignUtil.getSign(param, payConfig.getKey());
        param.put("sign", sign);
    
        String result = "";
        Map<String, Object> map = null;
        try {
            String requestData = XmlUtil.mapToXml(param);
            URL url = new URL(payConfig.getFilePath());
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            conn.setRequestMethod("GET");
            conn.setConnectTimeout(5 * 1000);
            InputStream inputStream = conn.getInputStream();
            logger.info("退款请求参数：{}", requestData);
            result = requestOnce(requestData, payConfig, inputStream,true, refundUrl);
            map = XmlUtil.xmlToMap(result);
        } catch (Exception e) {
            e.printStackTrace();
        }
        if (map == null) {
            logger.info("请求失败");
            return;
        }
        Refund refund = new Refund();
        refund.setCreateTime(new Date());
        refund.setResultStr(result);
        refund.setRefundNo(refundNo);
        refund.setOrderId(order.getId());
        refund.setResultStr(result);
    
        logger.info("退款请求返回参数：{}", result);
        if ("FAIL".equals(map.get("return_code"))) {
            logger.info("退款失败");
            refund.setStatus("FAIL");
            refundService.addRefund(refund);
    
            order.setOrderStatus("REFUND_FAIL");
            order.setStatus("OVERDATE");
            activityOrderService.updateActivityOrder(order);
            return;
        }
    
        if ("FAIL".equals(map.get("result_code"))) {
            logger.info("退款失败");
            refund.setStatus("FAIL");
            refundService.addRefund(refund);
    
            order.setOrderStatus("REFUND_FAIL");
            order.setStatus("OVERDATE");
            activityOrderService.updateActivityOrder(order);
            return;
        }
        refund.setStatus("SUCCESS");
        refund.setOrderNo((String) map.get("out_trade_no"));
        refund.setWechatNo((String)map.get("transaction_id"));
        refundService.addRefund(refund);
    
    
        order.setOrderStatus("REFUND");
        order.setStatus("OVERDATE");
        activityOrderService.updateActivityOrder(order);
    }

最开始本人使用的是定时任务

/**
         * 处理拼团时间到期
         */
    //    @Async
    //    @Scheduled(cron = "0 0/1 * * * *")
        public void updatePinTuanEnd(){
            Date date=new Date();
           List<Activity> activityList=activityService.queryStarting(date);
           List<GroupDataBase> activityOrderList=null;
           List<ActivityOrder> orders=null;
           List<ActivityOrder> groupOrder=null;
           Map<String,GroupDataBase> map=null;
           Date pintuanTime=null;
           for(Activity activity:activityList){
               if(activity.getTimeLimit()){//设置拼团时间限制
                   activityOrderList=activityOrderService.queryGroupData(activity.getId());
                   map=activityOrderList.stream().collect(Collectors.toMap(GroupDataBase::getGroupNo, Function.identity(), (key1, key2) -> key2));
                   orders=activityOrderService.queryByActivityIdAndIsMain(activity.getId());
                   if(orders !=null && orders.size()>0){
                       for(ActivityOrder order:orders){
                           pintuanTime=TimeUtils.addDate(order.getPayTime(),activity.getDayLimit(),activity.getHourLimit(),activity.getMinuteLimit());
                           if(new Date().after(pintuanTime)){
                               if(!activity.getAutoGroup()){
                                   logger.info("activityId={}",activity.getId());
                                   if(map.get(order.getGroupNo()).getAmount()<activity.getMinAmount()){
                                       groupOrder=activityOrderService.queryByActivityIdAndGroupNo(activity.getId(),order.getGroupNo());
                                       for(ActivityOrder order1:groupOrder){
                                           order1.setOrderStatus(ActivityOrderStatusEnum.REFUND.getName());
                                           order1.setStatus(ActivityOrderServiceEnum.OVERDATE.getName());
                                           activityOrderService.updateActivityOrder(order1);
                                       }
    
                                   }
                               }
    
                           }
    
                       }
                   }
               }
    
           }
        }

定时任务也可以实现

定时任务与消息队列的区别

1.定时任务很消耗系统内存

2.定时任务倾向于批处理，消息队列倾向于逐条处理