From f4439e5c8bdb556fd0c204c11bf8e1634506a022 Mon Sep 17 00:00:00 2001 From: luozukai Date: Thu, 12 Sep 2019 20:20:45 +0800 Subject: [PATCH] =?UTF-8?q?[=E4=BF=AE=E6=94=B9][=E7=BB=93=E7=AE=97?= =?UTF-8?q?=E5=8D=95][=E4=BF=AE=E6=94=B9=E7=BB=93=E7=AE=97=E5=8D=95]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../schedule/OrderExpiringSchedule.java | 59 +++++++++++++++++-- .../mq/impl/RabbitMqMessageProducer.java | 8 +-- 2 files changed, 57 insertions(+), 10 deletions(-) diff --git a/mallinkSchedule/src/main/java/com/iformall/schedule/OrderExpiringSchedule.java b/mallinkSchedule/src/main/java/com/iformall/schedule/OrderExpiringSchedule.java index b362e5a7d..230872b9e 100644 --- a/mallinkSchedule/src/main/java/com/iformall/schedule/OrderExpiringSchedule.java +++ b/mallinkSchedule/src/main/java/com/iformall/schedule/OrderExpiringSchedule.java @@ -2,17 +2,18 @@ package com.iformall.schedule; import com.iformall.domain.po.WxAppinfo; import com.iformall.domain.po.WxOrder; -import com.iformall.enums.EnumAppType; -import com.iformall.enums.EnumOrderStatus; -import com.iformall.enums.EnumOrderType; -import com.iformall.enums.EnumRefundStatus; +import com.iformall.domain.po.msg.*; +import com.iformall.enums.*; import com.iformall.mapper.WxAppinfoMapper; import com.iformall.mapper.WxCouponMapper; import com.iformall.mapper.WxOrderGroupMapper; import com.iformall.mapper.WxOrderMapper; +import com.iformall.mq.MqBaseProducer; +import com.iformall.service.WxMsgRecordService; import com.iformall.service.WxOrderGroupService; import com.iformall.service.WxOrderService; import com.iformall.service.WxRefundOrderService; +import com.iformall.utils.JsonUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -54,6 +55,12 @@ public class OrderExpiringSchedule { @Autowired WxCouponMapper wxCouponMapper; + @Autowired + WxMsgRecordService wxMsgRecordService; + + @Autowired + MqBaseProducer mqBaseProducer; + @Scheduled(cron = "0 */5 * * * *?") // 每5分钟检查一次 //@Scheduled(cron = "*/10 * * * * ?") // 测试10秒中一次 public void orderExpireSchedule() { @@ -143,7 +150,49 @@ public class OrderExpiringSchedule { } }); - + //消息重发 + WxMsgRecord msgRecord = new WxMsgRecord(); + msgRecord.setMsgStatus(EnumMsgRecordStatus.SEND_FAIL.getCode()); + List recordList = wxMsgRecordService.findList(msgRecord); + for (WxMsgRecord wxMsgRecord:recordList) { + logger.info("消息重发:{}",wxMsgRecord.getMsgJson()); + try { + if(EnumMsgRecordType.INSIDE_ORDER_SUCCESS.getCode().equals(wxMsgRecord.getMsgType())){ + //内部消息 - 下订单成功 + FmInsideOrderSuccessMsg msg = (FmInsideOrderSuccessMsg) JsonUtil.readValue(wxMsgRecord.getMsgJson(),FmInsideOrderSuccessMsg.class); + msg.setMsgType(wxMsgRecord.getMsgType()); + msg.setReceiver(wxMsgRecord.getReceiver()); + msg.setTenantId(wxMsgRecord.getTenantId()); + mqBaseProducer.sendMessage(msg, EnumMsgMqTopic.DEFAULT.getCode(), EnumMsgMqTag.DEFAULT.getCode(), EnumMsgMqKey.DEFAULT.getCode()); + }else if(EnumMsgRecordType.INSIDE_COUPON_VERIFY.getCode().equals(wxMsgRecord.getMsgType())){ + FmInsideCouponVerifyMsg msg = (FmInsideCouponVerifyMsg) JsonUtil.readValue(wxMsgRecord.getMsgJson(),FmInsideCouponVerifyMsg.class); + msg.setMsgType(wxMsgRecord.getMsgType()); + msg.setReceiver(wxMsgRecord.getReceiver()); + msg.setTenantId(wxMsgRecord.getTenantId()); + mqBaseProducer.sendMessage(msg, EnumMsgMqTopic.DEFAULT.getCode(), EnumMsgMqTag.DEFAULT.getCode(), EnumMsgMqKey.DEFAULT.getCode()); + }else if(EnumMsgRecordType.INSIDE_C_LOGIN.getCode().equals(wxMsgRecord.getMsgType())){ + FmInsideCLoginMsg msg = (FmInsideCLoginMsg) JsonUtil.readValue(wxMsgRecord.getMsgJson(),FmInsideCLoginMsg.class); + msg.setMsgType(wxMsgRecord.getMsgType()); + msg.setReceiver(wxMsgRecord.getReceiver()); + msg.setTenantId(wxMsgRecord.getTenantId()); + mqBaseProducer.sendMessage(msg, EnumMsgMqTopic.DEFAULT.getCode(), EnumMsgMqTag.DEFAULT.getCode(), EnumMsgMqKey.DEFAULT.getCode()); + }else if(EnumMsgRecordType.INSIDE_NOTIFY_PAY_SUCCESS.getCode().equals(wxMsgRecord.getMsgType())){ + FmInsideNotifyPaySuccessMsg msg = (FmInsideNotifyPaySuccessMsg) JsonUtil.readValue(wxMsgRecord.getMsgJson(),FmInsideNotifyPaySuccessMsg.class); + msg.setMsgType(wxMsgRecord.getMsgType()); + msg.setReceiver(wxMsgRecord.getReceiver()); + msg.setTenantId(wxMsgRecord.getTenantId()); + mqBaseProducer.sendMessage(msg, EnumMsgMqTopic.DEFAULT.getCode(), EnumMsgMqTag.DEFAULT.getCode(), EnumMsgMqKey.DEFAULT.getCode()); + }else if(EnumMsgRecordType.INSIDE_NOTIFY_REFUND_SUCCESS.getCode().equals(wxMsgRecord.getMsgType())){ + FmInsideNotifyRefundSuccessMsg msg = (FmInsideNotifyRefundSuccessMsg) JsonUtil.readValue(wxMsgRecord.getMsgJson(),FmInsideNotifyRefundSuccessMsg.class); + msg.setMsgType(wxMsgRecord.getMsgType()); + msg.setReceiver(wxMsgRecord.getReceiver()); + msg.setTenantId(wxMsgRecord.getTenantId()); + mqBaseProducer.sendMessage(msg, EnumMsgMqTopic.DEFAULT.getCode(), EnumMsgMqTag.DEFAULT.getCode(), EnumMsgMqKey.DEFAULT.getCode()); + } + } catch (Exception e) { + logger.error("消息重发错误:" + e.getMessage()); + } + } } @Transactional(propagation = Propagation.REQUIRED, readOnly = false, rollbackFor = {Exception.class}) diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java b/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java index f248ec48c..1123a9547 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java +++ b/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java @@ -46,17 +46,15 @@ public class RabbitMqMessageProducer implements MqBaseProducer , RabbitTemplate. @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { log.info("confirm:: correlationData {},ack {}, cause {}", correlationData,ack,cause); - System.out.println(); - if(ack){ - System.out.println(); - }else{ + if(!ack){ BaseMsg data = new BaseMsg(); + data.setUuid(correlationData.getId()); data.setMsgStatus(EnumMsgRecordStatus.SEND_FAIL.getCode()); + data.setStatusMessage(cause); wxMsgRecordMapper.update(data); } } - @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { log.info("returnedMessage::message {} 消息发送失败,应答码:{}, 应答描述:{} ,exchange: {} ,routingKey :{}", new String(message.getBody()), replyCode, replyText, exchange, routingKey);