mq 重构：默认的 mq 实现，采用 event，减少对 redis、rocketmq 等消息队列的依赖

YunaiV
1 parent 87fb47ca
Showing 19 changed files with 68 additions and 87 deletions
yudao-module-mall/yudao-module-promotion-biz/src/main/java/cn/iocoder/yudao/module/promotion/mq/consumer/package-info.java
yudao-module-mall/yudao-module-promotion-biz/src/main/java/cn/iocoder/yudao/module/promotion/mq/message/package-info.java
yudao-module-mall/yudao-module-promotion-biz/src/main/java/cn/iocoder/yudao/module/promotion/mq/producer/package-info.java
yudao-module-member/yudao-module-member-api/src/main/java/cn/iocoder/yudao/module/member/message/package-info.java
yudao-module-member/yudao-module-member-biz/src/main/java/cn/iocoder/yudao/module/member/controller/app/auth/AppAuthController.http
yudao-module-member/yudao-module-member-biz/src/main/java/cn/iocoder/yudao/module/member/mq/consumer/package-info.java
yudao-module-member/yudao-module-member-biz/src/main/java/cn/iocoder/yudao/module/member/mq/message/package-info.java
yudao-module-member/yudao-module-member-biz/src/main/java/cn/iocoder/yudao/module/member/mq/message/user/MemberUserCreateMessage.java
yudao-module-member/yudao-module-member-biz/src/main/java/cn/iocoder/yudao/module/member/mq/producer/package-info.java
yudao-module-member/yudao-module-member-biz/src/main/java/cn/iocoder/yudao/module/member/mq/producer/user/MemberUserProducer.java
yudao-module-system/yudao-module-system-biz/src/main/java/cn/iocoder/yudao/module/system/controller/admin/sms/SmsTemplateController.http
yudao-module-system/yudao-module-system-biz/src/main/java/cn/iocoder/yudao/module/system/mq/consumer/mail/MailSendConsumer.java
yudao-module-system/yudao-module-system-biz/src/main/java/cn/iocoder/yudao/module/system/mq/consumer/sms/SmsSendConsumer.java
yudao-module-system/yudao-module-system-biz/src/main/java/cn/iocoder/yudao/module/system/mq/message/mail/MailSendMessage.java
yudao-module-system/yudao-module-system-biz/src/main/java/cn/iocoder/yudao/module/system/mq/message/sms/SmsSendMessage.java
yudao-module-system/yudao-module-system-biz/src/main/java/cn/iocoder/yudao/module/system/mq/producer/mail/MailProducer.java
yudao-module-system/yudao-module-system-biz/src/main/java/cn/iocoder/yudao/module/system/mq/producer/sms/SmsProducer.java
 package cn.iocoder.yudao.module.promotion.mq.consumer.coupon;
-import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessageListener;
-import cn.iocoder.yudao.module.promotion.mq.message.coupon.UserCreateMessage;
+import cn.iocoder.yudao.module.member.message.user.MemberUserCreateMessage;
 import cn.iocoder.yudao.module.promotion.service.coupon.CouponService;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 import javax.annotation.Resource;
 /**
- * 针对 {@link UserCreateMessage} 的消费者
+ * 用户注册时，发送优惠劵的消费者，基 {@link MemberUserCreateMessage} 消息
  *
  * @author owen
  */
 @Component
 @Slf4j
-public class UserCreateConsumer extends AbstractStreamMessageListener<UserCreateMessage> {
+public class CouponTakeByRegisterConsumer {
     @Resource
     private CouponService couponService;
-    @Override
-    public void onMessage(UserCreateMessage message) {
+    @EventListener
+    @Async // Spring Event 默认在 Producer 发送的线程，通过 @Async 实现异步
+    public void onMessage(MemberUserCreateMessage message) {
         log.info("[onMessage][消息内容({})]", message);
         couponService.takeCouponByRegister(message.getUserId());
     }
+/**
+ * 消息队列的消费者
+ */
+package cn.iocoder.yudao.module.promotion.mq.consumer;
+/**
+ * 消息队列的消息
+ */
+package cn.iocoder.yudao.module.promotion.mq.message;
+/**
+ * 消息队列的生产者
+ */
+package cn.iocoder.yudao.module.promotion.mq.producer;
+/**
+ * 消息队列的消息
+ */
+package cn.iocoder.yudao.module.member.message;
-package cn.iocoder.yudao.module.promotion.mq.message.coupon;
+package cn.iocoder.yudao.module.member.message.user;
-import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessage;
 import lombok.Data;
-import lombok.EqualsAndHashCode;
 import javax.validation.constraints.NotNull;
@@ -12,8 +10,7 @@ import javax.validation.constraints.NotNull;
  * @author owen
  */
 @Data
-@EqualsAndHashCode(callSuper = true)
-public class UserCreateMessage extends AbstractStreamMessage {
+public class MemberUserCreateMessage {
     /**
      * 用户编号
@@ -21,9 +18,4 @@ public class UserCreateMessage extends AbstractStreamMessage {
     @NotNull(message = "用户编号不能为空")
     private Long userId;
-    @Override
-    public String getStreamKey() {
-        return "member.user.create";
-    }
-
 }
@@ -4,7 +4,7 @@ Content-Type: application/json
 tenant-id: {{appTenentId}}
 {
-  "mobile": "15601691300",
+  "mobile": "15601691388",
   "password": "admin123"
 }
@@ -14,7 +14,7 @@ Content-Type: application/json
 tenant-id: {{appTenentId}}
 {
-  "mobile": "15601691399",
+  "mobile": "15601691388",
   "scene": 1
 }
@@ -22,9 +22,10 @@ tenant-id: {{appTenentId}}
 POST {{appApi}}/member/auth/sms-login
 Content-Type: application/json
 tenant-id: {{appTenentId}}
+terminal: 30
 {
-  "mobile": "15601691301",
+  "mobile": "15601691388",
   "code": 9999
 }
+/**
+ * 消息队列的消费者
+ */
+package cn.iocoder.yudao.module.member.mq.consumer;
+/**
+ * 消息队列的消息
+ */
+package cn.iocoder.yudao.module.member.mq.message;
-package cn.iocoder.yudao.module.member.mq.message.user;
-
-import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessage;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-
-import javax.validation.constraints.NotNull;
-
-/**
- * 会员用户创建消息
- *
- * @author owen
- */
-@Data
-@EqualsAndHashCode(callSuper = true)
-public class MemberUserCreateMessage extends AbstractStreamMessage {
-
-    /**
-     * 用户编号
-     */
-    @NotNull(message = "用户编号不能为空")
-    private Long userId;
-
-    @Override
-    public String getStreamKey() {
-        return "member.user.create";
-    }
-
-}
+/**
+ * 消息队列的生产者
+ */
+package cn.iocoder.yudao.module.member.mq.producer;
 package cn.iocoder.yudao.module.member.mq.producer.user;
-import cn.iocoder.yudao.framework.mq.core.RedisMQTemplate;
-import cn.iocoder.yudao.module.member.mq.message.user.MemberUserCreateMessage;
+import cn.iocoder.yudao.module.member.message.user.MemberUserCreateMessage;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationContext;
 import org.springframework.stereotype.Component;
 import javax.annotation.Resource;
@@ -17,7 +17,7 @@ import javax.annotation.Resource;
 public class MemberUserProducer {
     @Resource
-    private RedisMQTemplate redisMQTemplate;
+    private ApplicationContext applicationContext;
     /**
      * 发送 {@link MemberUserCreateMessage} 消息
@@ -25,7 +25,7 @@ public class MemberUserProducer {
      * @param userId 用户编号
      */
     public void sendUserCreateMessage(Long userId) {
-        redisMQTemplate.send(new MemberUserCreateMessage().setUserId(userId));
+        applicationContext.publishEvent(new MemberUserCreateMessage().setUserId(userId));
     }
 }
@@ -6,9 +6,9 @@ tenant-id: {{adminTenentId}}
 {
   "templateCode": "test_01",
-  "mobile": "156016913900",
-  "params": {
-    "key01": "value01",
-    "key02": "value02"
+  "mobile": "15601691390",
+  "templateParams": {
+    "operation": "value01",
+    "code": "value02"
   }
 }
 package cn.iocoder.yudao.module.system.mq.consumer.mail;
-import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessageListener;
 import cn.iocoder.yudao.module.system.mq.message.mail.MailSendMessage;
-import cn.iocoder.yudao.module.system.mq.message.sms.SmsSendMessage;
 import cn.iocoder.yudao.module.system.service.mail.MailSendService;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 import javax.annotation.Resource;
@@ -16,12 +16,13 @@ import javax.annotation.Resource;
  */
 @Component
 @Slf4j
-public class MailSendConsumer extends AbstractStreamMessageListener<MailSendMessage> {
+public class MailSendConsumer {
     @Resource
     private MailSendService mailSendService;
-    @Override
+    @EventListener
+    @Async // Spring Event 默认在 Producer 发送的线程，通过 @Async 实现异步
     public void onMessage(MailSendMessage message) {
         log.info("[onMessage][消息内容({})]", message);
         mailSendService.doSendMail(message);
@@ -2,8 +2,9 @@ package cn.iocoder.yudao.module.system.mq.consumer.sms;
 import cn.iocoder.yudao.module.system.mq.message.sms.SmsSendMessage;
 import cn.iocoder.yudao.module.system.service.sms.SmsSendService;
-import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessageListener;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 import javax.annotation.Resource;
@@ -15,12 +16,13 @@ import javax.annotation.Resource;
  */
 @Component
 @Slf4j
-public class SmsSendConsumer extends AbstractStreamMessageListener<SmsSendMessage> {
+public class SmsSendConsumer {
     @Resource
     private SmsSendService smsSendService;
-    @Override
+    @EventListener
+    @Async // Spring Event 默认在 Producer 发送的线程，通过 @Async 实现异步
     public void onMessage(SmsSendMessage message) {
         log.info("[onMessage][消息内容({})]", message);
         smsSendService.doSendSms(message);
 package cn.iocoder.yudao.module.system.mq.message.mail;
-import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessage;
 import lombok.Data;
-import lombok.EqualsAndHashCode;
 import javax.validation.constraints.NotEmpty;
 import javax.validation.constraints.NotNull;
@@ -13,8 +11,7 @@ import javax.validation.constraints.NotNull;
  * @author 芋道源码
  */
 @Data
-@EqualsAndHashCode(callSuper = true)
-public class MailSendMessage extends AbstractStreamMessage {
+public class MailSendMessage {
     /**
      * 邮件日志编号
@@ -47,9 +44,4 @@ public class MailSendMessage extends AbstractStreamMessage {
     @NotEmpty(message = "邮件内容不能为空")
     private String content;
-    @Override
-    public String getStreamKey() {
-        return "system.mail.send";
-    }
-
 }
 package cn.iocoder.yudao.module.system.mq.message.sms;
 import cn.iocoder.yudao.framework.common.core.KeyValue;
-import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessage;
 import lombok.Data;
-import lombok.EqualsAndHashCode;
 import javax.validation.constraints.NotNull;
 import java.util.List;
@@ -14,8 +12,7 @@ import java.util.List;
  * @author 芋道源码
  */
 @Data
-@EqualsAndHashCode(callSuper = true)
-public class SmsSendMessage extends AbstractStreamMessage {
+public class SmsSendMessage {
     /**
      * 短信日志编号
@@ -42,9 +39,4 @@ public class SmsSendMessage extends AbstractStreamMessage {
      */
     private List<KeyValue<String, Object>> templateParams;
-    @Override
-    public String getStreamKey() {
-        return "system.sms.send";
-    }
-
 }
 package cn.iocoder.yudao.module.system.mq.producer.mail;
-import cn.iocoder.yudao.framework.mq.core.RedisMQTemplate;
 import cn.iocoder.yudao.module.system.mq.message.mail.MailSendMessage;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationContext;
 import org.springframework.stereotype.Component;
 import javax.annotation.Resource;
@@ -18,7 +18,7 @@ import javax.annotation.Resource;
 public class MailProducer {
     @Resource
-    private RedisMQTemplate redisMQTemplate;
+    private ApplicationContext applicationContext;
     /**
      * 发送 {@link MailSendMessage} 消息
@@ -35,7 +35,7 @@ public class MailProducer {
         MailSendMessage message = new MailSendMessage()
                 .setLogId(sendLogId).setMail(mail).setAccountId(accountId)
                 .setNickname(nickname).setTitle(title).setContent(content);
-        redisMQTemplate.send(message);
+        applicationContext.publishEvent(message);
     }
 }
 package cn.iocoder.yudao.module.system.mq.producer.sms;
 import cn.iocoder.yudao.framework.common.core.KeyValue;
-import cn.iocoder.yudao.framework.mq.core.RedisMQTemplate;
 import cn.iocoder.yudao.module.system.mq.message.sms.SmsSendMessage;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationContext;
 import org.springframework.stereotype.Component;
 import javax.annotation.Resource;
@@ -20,7 +20,7 @@ import java.util.List;
 public class SmsProducer {
     @Resource
-    private RedisMQTemplate redisMQTemplate;
+    private ApplicationContext applicationContext;
     /**
      * 发送 {@link SmsSendMessage} 消息
@@ -35,7 +35,7 @@ public class SmsProducer {
                                    Long channelId, String apiTemplateId, List<KeyValue<String, Object>> templateParams) {
         SmsSendMessage message = new SmsSendMessage().setLogId(logId).setMobile(mobile);
         message.setChannelId(channelId).setApiTemplateId(apiTemplateId).setTemplateParams(templateParams);
-        redisMQTemplate.send(message);
+        applicationContext.publishEvent(message);
     }
 }