message消费防重

sckw-common/sckw-common-stream/src/main/java/com/sckw/stream/model/SckwMessage.java

@@ -7,12 +7,21 @@ import lombok.NoArgsConstructor;
 
 import java.util.Date;
 import java.util.Map;
-
+/**
+ * @desc: message消费对象
+ * @author: yzc
+ * @date: 2023-06-19 13:52
+ */
 @Data
 @NoArgsConstructor
 @AllArgsConstructor
 public class SckwMessage {
 
+    /**
+     * 请求id
+     */
+    private String requestId;
+
     /**
      * 短信模板code
      */

sckw-modules/sckw-message/src/main/java/com/sckw/message/constant/RedisConstant.java

@@ -10,9 +10,20 @@ public class RedisConstant {
     /**
      * 短信验证码key
      */
-    public static final String MESSAGE_SMS_VERIFY_CODE_KEY = "sckw:message:sms:verifyCode:%s";
+    public static final String MESSAGE_SMS_VERIFY_CODE_KEY = "sckw:sms:verifyCode:%s";
     /**
      * 短信验证码有效时间（秒）
      */
     public static final Long SMS_VERIFY_CODE_VALID_TIME = 180L;
+
+    /**
+     * message消费请求key
+     */
+    public static final String MESSAGE_CONSUMER_REQUEST_KEY = "sckw:message:consumer:request:%s";
+
+    /**
+     * 消费请求有效时间（秒）
+     */
+    public static final Long CONSUMER_REQUEST_VALID_TIME = 180L;
+
 }

sckw-modules/sckw-message/src/main/java/com/sckw/message/service/SckwMessageHandlerService.java

@@ -3,8 +3,10 @@ package com.sckw.message.service;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
 import com.sckw.core.utils.StringUtils;
+import com.sckw.message.constant.RedisConstant;
 import com.sckw.message.model.KwmMessage;
 import com.sckw.message.model.KwmMessageUser;
+import com.sckw.redis.utils.RedissonUtils;
 import com.sckw.stream.enums.MessageEnum;
 import com.sckw.stream.model.SckwMessage;
 import lombok.RequiredArgsConstructor;
@@ -29,15 +31,19 @@ public class SckwMessageHandlerService {
     private final KwmMessageUserService kwmMessageUserService;
 
 
-    /**
-     * @param sckwMessage
-     * @return void
-     * @desc: message消息处理
-     * @author: yzc
-     * @date: 2023-06-09 15:33
-     */
-    @Transactional(rollbackFor = Exception.class)
+/**
+ * @desc: message消息处理
+ * @author: yzc
+ * @date: 2023-06-09 15:33
+ * @Param sckwMessage:
+ * @return: void
+ */
+@Transactional(rollbackFor = Exception.class)
     public void handler(SckwMessage sckwMessage) {
+        if (Boolean.FALSE.equals(RedissonUtils.tryLock(getKey(sckwMessage.getRequestId()), 10L, RedisConstant.CONSUMER_REQUEST_VALID_TIME))) {
+            log.info("重复message消息：{}，不处理", sckwMessage.getRequestId());
+            return;
+        }
         String userIds = sckwMessage.getUserIds();
         if (StringUtils.isBlank(userIds)) {
             log.error("message消息处理用户id为空，不处理");
@@ -62,4 +68,8 @@ public class SckwMessageHandlerService {
         });
         kwmMessageUserService.batchSave(messageUsers);
     }
+
+    public static String getKey(String requestId) {
+        return String.format(RedisConstant.MESSAGE_CONSUMER_REQUEST_KEY, requestId);
+    }
 }