From 1d8cb26e117b9dabee3cc8b51cf34aa82f4e181d Mon Sep 17 00:00:00 2001 From: luozukai Date: Fri, 8 Mar 2019 18:45:59 +0800 Subject: [PATCH] =?UTF-8?q?[=E6=B6=88=E6=81=AF=E7=BB=84=E4=BB=B6][?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0][=E6=B7=BB=E5=8A=A0=E5=BE=AE=E4=BF=A1?= =?UTF-8?q?=E5=B0=8F=E7=A8=8B=E5=BA=8F=E9=80=9A=E7=9F=A5]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iformall/controller/HomeController.java | 17 +++++++++++++++- .../java/com/iformall/mq/MqBaseConsumer.java | 20 ++++++++++++++----- .../iformall/mq/impl/RocketMqConsumer.java | 2 +- .../java/com/iformall/domain/po/BaseMsg.java | 10 ++++++++++ .../com/iformall/domain/po/SmartAppMsg.java | 1 + .../mq/impl/RocketMqMessageProducer.java | 3 +++ 6 files changed, 46 insertions(+), 7 deletions(-) diff --git a/mallinkMQConsumer/src/main/java/com/iformall/controller/HomeController.java b/mallinkMQConsumer/src/main/java/com/iformall/controller/HomeController.java index d7d440b98..2f6d914c6 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/controller/HomeController.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/controller/HomeController.java @@ -39,7 +39,6 @@ public class HomeController { @Value("${spring.mail.username}") private String from; - @ApiOperation("获取后端版本号") @GetMapping("/version") public ResultData version() { @@ -75,4 +74,20 @@ public class HomeController { mqBaseProducer.sendMessage(wxMsgRecord, "topic-1", "tag-1", "key-1"); return new ResultData(); } + + /** + * 重发 + * @return + */ + @GetMapping("/resend") + public ResultData resend() { + MailMsg wxMsgRecord = new MailMsg(); + wxMsgRecord.setMsgType(EnumMsgRecordType.EMAIL.getCode()); + wxMsgRecord.setFrom(from); + wxMsgRecord.setTo(new String[]{"luozukai@iformall.com"}); + wxMsgRecord.setSubject("回家吃饭了"); + wxMsgRecord.setMsg("大熊熊"); + mqBaseProducer.sendMessage(wxMsgRecord, "topic-1", "tag-1", "key-1"); + return new ResultData(); + } } \ No newline at end of file diff --git a/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java b/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java index dc32f4ba7..7e0d5495a 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java @@ -1,8 +1,7 @@ package com.iformall.mq; import com.iformall.domain.po.*; -import com.iformall.enums.EnumMsgRecordStatus; -import com.iformall.enums.EnumMsgRecordType; +import com.iformall.enums.*; import com.iformall.mapper.WxMsgRecordMapper; import com.iformall.service.impl.SendCallBackSmsServiceImpl; import com.iformall.service.impl.SendEmailServiceImpl; @@ -30,6 +29,8 @@ public class MqBaseConsumer { private SendEmailServiceImpl sendEmailService; @Autowired private SendSmartAppMsgServiceImpl sendSmartAppMsgService; + @Autowired + private MqBaseProducer mqBaseProducer; public void doMessage(String message) { log.info("received message: {}", message); @@ -39,12 +40,16 @@ public class MqBaseConsumer { baseMsg = (BaseMsg)JsonUtil.readValue(message,BaseMsg.class); if(EnumMsgRecordType.SMS.getCode().equals(baseMsg.getMsgType())){ + //短信 sendSmsService.send((WxMsgRecord)JsonUtil.readValue(message,WxMsgRecord.class)); }else if(EnumMsgRecordType.SMS_CALLBACK.getCode().equals(baseMsg.getMsgType())){ + //业务短信 sendCallBackSmsService.send((WxMsg)JsonUtil.readValue(message,WxMsg.class)); }else if(EnumMsgRecordType.EMAIL.getCode().equals(baseMsg.getMsgType())){ + //邮件 sendEmailService.send((MailMsg)JsonUtil.readValue(message,MailMsg.class)); }else if(EnumMsgRecordType.SMART_APP.getCode().equals(baseMsg.getMsgType())){ + //微信小程序 sendSmartAppMsgService.send((SmartAppMsg)JsonUtil.readValue(message,SmartAppMsg.class)); } @@ -52,9 +57,14 @@ public class MqBaseConsumer { wxMsgRecordMapper.update(baseMsg); }catch (Exception e){ if(null != baseMsg){ - baseMsg.setStatusMessage(e.getMessage()); - baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_FAIL.getCode()); - wxMsgRecordMapper.update(baseMsg); + if(baseMsg.getDelayTimeLevel() < 3){ + baseMsg.setDelayTimeLevel(3); //10s后重试 + mqBaseProducer.sendMessage(baseMsg, EnumMsgMqTopic.DEFAULT.getCode(), EnumMsgMqTag.DEFAULT.getCode(), EnumMsgMqKey.DEFAULT.getCode()); + }else{ + baseMsg.setStatusMessage(e.getMessage()); + baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_FAIL.getCode()); + wxMsgRecordMapper.update(baseMsg); + } } } diff --git a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java index 0508f1df7..ee1d0681d 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java @@ -9,7 +9,7 @@ import org.springframework.context.annotation.Profile; @Slf4j @Profile(MQConfig.Impl.ROCKET_MQ) -@RocketMQListener(topic = "topic-1") +@RocketMQListener(topic = "topic-1",consumeThreadMax=100,consumeThreadMin = 20) public class RocketMqConsumer extends MqBaseConsumer { @RocketMQMessage(messageClass = String.class, tag = "tag-1") diff --git a/mallinkService/src/main/java/com/iformall/domain/po/BaseMsg.java b/mallinkService/src/main/java/com/iformall/domain/po/BaseMsg.java index cbc92964b..b250edb16 100644 --- a/mallinkService/src/main/java/com/iformall/domain/po/BaseMsg.java +++ b/mallinkService/src/main/java/com/iformall/domain/po/BaseMsg.java @@ -21,6 +21,16 @@ public class BaseMsg implements Serializable { //结果信息 private String statusMessage; + //延迟消息1s,5s,10s,30s,1m,2m,3m,4m,5m,6m,7m,8m,9m,10m,20m,30m,1h,2h + private Integer delayTimeLevel = 0; + + public Integer getDelayTimeLevel() { + return delayTimeLevel; + } + + public void setDelayTimeLevel(Integer delayTimeLevel) { + this.delayTimeLevel = delayTimeLevel; + } public String getStatusMessage() { return statusMessage; diff --git a/mallinkService/src/main/java/com/iformall/domain/po/SmartAppMsg.java b/mallinkService/src/main/java/com/iformall/domain/po/SmartAppMsg.java index 14e938ba9..cddc63e09 100644 --- a/mallinkService/src/main/java/com/iformall/domain/po/SmartAppMsg.java +++ b/mallinkService/src/main/java/com/iformall/domain/po/SmartAppMsg.java @@ -4,6 +4,7 @@ import cn.binarywang.wx.miniapp.bean.WxMaTemplateData; import java.util.List; /** + * 微信小程序消息 * @author luozukai * @date 2019/3/7 10:05 */ diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java b/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java index 1feb665be..13db5f4f4 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java +++ b/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java @@ -64,6 +64,9 @@ public class RocketMqMessageProducer implements MqBaseProducer { byte[] messageBody = JsonUtil.obj2Json(data).getBytes(RemotingHelper.DEFAULT_CHARSET); Message mqMsg = new Message(topic, tags, keys, messageBody); + if(data.getDelayTimeLevel() > 0){ + mqMsg.setDelayTimeLevel(data.getDelayTimeLevel()); + } producer.send(mqMsg, new SendCallback() { @Override