diff --git a/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java b/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java index d9a0b3ff3..253a2a270 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java @@ -5,7 +5,6 @@ import com.iformall.enums.*; import com.iformall.mapper.WxMsgRecordMapper; import com.iformall.service.msg.impl.*; import com.iformall.utils.JsonUtil; -import org.checkerframework.checker.units.qual.A; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; diff --git a/mallinkSchedule/src/main/java/com/iformall/schedule/MsgReSendSchedule.java b/mallinkSchedule/src/main/java/com/iformall/schedule/MsgReSendSchedule.java new file mode 100644 index 000000000..ac009a214 --- /dev/null +++ b/mallinkSchedule/src/main/java/com/iformall/schedule/MsgReSendSchedule.java @@ -0,0 +1,77 @@ +package com.iformall.schedule; + +import com.iformall.domain.po.msg.*; +import com.iformall.enums.*; +import com.iformall.mq.MqBaseProducer; +import com.iformall.service.WxMsgRecordService; +import com.iformall.utils.JsonUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import java.util.List; + +/** + * 消息重发 + */ +@Component +public class MsgReSendSchedule { + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + @Autowired + private WxMsgRecordService wxMsgRecordService; + @Autowired + private MqBaseProducer mqBaseProducer; + + @Scheduled(cron = "0 */5 * * * *?") // 每5分钟检查一次 + //@Scheduled(cron = "*/10 * * * * ?") // 测试10秒中一次 + public void msgReSend() { + //消息重发 + 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()); + } + } + + } + + +} \ No newline at end of file diff --git a/mallinkSchedule/src/main/java/com/iformall/schedule/OrderExpiringSchedule.java b/mallinkSchedule/src/main/java/com/iformall/schedule/OrderExpiringSchedule.java index 230872b9e..6af580675 100644 --- a/mallinkSchedule/src/main/java/com/iformall/schedule/OrderExpiringSchedule.java +++ b/mallinkSchedule/src/main/java/com/iformall/schedule/OrderExpiringSchedule.java @@ -2,7 +2,6 @@ package com.iformall.schedule; import com.iformall.domain.po.WxAppinfo; import com.iformall.domain.po.WxOrder; -import com.iformall.domain.po.msg.*; import com.iformall.enums.*; import com.iformall.mapper.WxAppinfoMapper; import com.iformall.mapper.WxCouponMapper; @@ -13,7 +12,6 @@ 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; @@ -21,7 +19,6 @@ import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; - import java.util.Date; import java.util.HashMap; import java.util.List; @@ -43,9 +40,6 @@ public class OrderExpiringSchedule { @Autowired WxOrderGroupMapper wxOrderGroupMapper; - @Autowired - WxOrderGroupService wxOrderGroupService; - @Autowired WxAppinfoMapper wxAppinfoMapper; @@ -55,12 +49,6 @@ public class OrderExpiringSchedule { @Autowired WxCouponMapper wxCouponMapper; - @Autowired - WxMsgRecordService wxMsgRecordService; - - @Autowired - MqBaseProducer mqBaseProducer; - @Scheduled(cron = "0 */5 * * * *?") // 每5分钟检查一次 //@Scheduled(cron = "*/10 * * * * ?") // 测试10秒中一次 public void orderExpireSchedule() { @@ -150,49 +138,6 @@ 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})