fix: 修复IoT使用redis消息队列时iot_device_message的STREAM的消息无限积压问题

panda
1 parent 3f1d86ef
Showing 4 changed files with 42 additions and 17 deletions
yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQConsumerAutoConfiguration.java
yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java
yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisStreamMessageCleanupJob.java
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java
@@ -12,7 +12,6 @@ import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessag
 import cn.iocoder.yudao.framework.redis.config.YudaoRedisAutoConfiguration;
 import lombok.extern.slf4j.Slf4j;
 import org.redisson.api.RedissonClient;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.context.annotation.Bean;
@@ -70,7 +69,8 @@ public class YudaoRedisMQConsumerAutoConfiguration {
     public RedisPendingMessageResendJob redisPendingMessageResendJob(List<AbstractRedisStreamMessageListener<?>> listeners,
                                                                      RedisMQTemplate redisTemplate,
                                                                      RedissonClient redissonClient) {
-        return new RedisPendingMessageResendJob(listeners, redisTemplate, redissonClient);
+        return new RedisPendingMessageResendJob(listeners, redisTemplate, redissonClient,
+                RedisPendingMessageResendJob.DEFAULT_RESEND_LOCK_KEY);
     }
     /**
@@ -81,7 +81,8 @@ public class YudaoRedisMQConsumerAutoConfiguration {
     public RedisStreamMessageCleanupJob redisStreamMessageCleanupJob(List<AbstractRedisStreamMessageListener<?>> listeners,
                                                                      RedisMQTemplate redisTemplate,
                                                                      RedissonClient redissonClient) {
-        return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient);
+        return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient,
+                RedisStreamMessageCleanupJob.DEFAULT_CLEANUP_LOCK_KEY);
     }
     /**
@@ -23,7 +23,9 @@ import java.util.Objects;
 @AllArgsConstructor
 public class RedisPendingMessageResendJob {
-    private static final String LOCK_KEY = "redis:stream:pending-message-resend:lock";
+    public static final String DEFAULT_RESEND_LOCK_KEY = "redis:stream:pending-message-resend:lock";
+
+    public static final String IOT_RESEND_LOCK_KEY = "redis:stream:pending-message-resend:lock:iot";
     /**
      * 消息超时时间，默认 5 分钟
@@ -36,22 +38,26 @@ public class RedisPendingMessageResendJob {
     private final List<AbstractRedisStreamMessageListener<?>> listeners;
     private final RedisMQTemplate redisTemplate;
     private final RedissonClient redissonClient;
+    private final String resendLockKey;
     /**
      * 一分钟执行一次,这里选择每分钟的 35 秒执行，是为了避免整点任务过多的问题
      */
     @Scheduled(cron = "35 * * * * ?")
     public void messageResend() {
-        RLock lock = redissonClient.getLock(LOCK_KEY);
-        // 尝试加锁
+        RLock lock = redissonClient.getLock(resendLockKey);
         if (lock.tryLock()) {
             try {
                 execute();
             } catch (Exception ex) {
-                log.error("[messageResend][执行异常]", ex);
+                log.error("[messageResend][执行异常][lockKey={}]", resendLockKey, ex);
             } finally {
-                lock.unlock();
+                if (lock.isHeldByCurrentThread()) {
+                    lock.unlock();
+                }
             }
+        } else {
+            log.debug("[messageResend][未获取到锁，跳过本轮][lockKey={}]", resendLockKey);
         }
     }
@@ -23,7 +23,16 @@ import java.util.List;
 @AllArgsConstructor
 public class RedisStreamMessageCleanupJob {
-    private static final String LOCK_KEY = "redis:stream:message-cleanup:lock";
+    /**
+     * 业务 MQ（Spring 容器内 AbstractRedisStreamMessageListener）清理任务使用的分布式锁
+     */
+    public static final String DEFAULT_CLEANUP_LOCK_KEY = "redis:stream:message-cleanup:lock";
+
+    /**
+     * IoT Redis 总线清理任务使用的分布式锁（须与 {@link #DEFAULT_CLEANUP_LOCK_KEY} 区分，否则会共抢一把锁，
+     * 同一时刻只有一侧能执行 XTRIM，另一侧 Stream 可能无限积压）
+     */
+    public static final String IOT_CLEANUP_LOCK_KEY = "redis:stream:message-cleanup:lock:iot";
     /**
      * 保留的消息数量，默认保留最近 10000 条消息
@@ -33,22 +42,29 @@ public class RedisStreamMessageCleanupJob {
     private final List<AbstractRedisStreamMessageListener<?>> listeners;
     private final RedisMQTemplate redisTemplate;
     private final RedissonClient redissonClient;
+    /**
+     * Redisson 锁键（多 Bean 注册清理任务时必须各不相同）
+     */
+    private final String cleanupLockKey;
     /**
      * 每小时执行一次清理任务
      */
     @Scheduled(cron = "0 0 * * * ?")
     public void cleanup() {
-        RLock lock = redissonClient.getLock(LOCK_KEY);
-        // 尝试加锁
+        RLock lock = redissonClient.getLock(cleanupLockKey);
         if (lock.tryLock()) {
             try {
                 execute();
             } catch (Exception ex) {
-                log.error("[cleanup][执行异常]", ex);
+                log.error("[cleanup][执行异常][lockKey={}]", cleanupLockKey, ex);
             } finally {
-                lock.unlock();
+                if (lock.isHeldByCurrentThread()) {
+                    lock.unlock();
+                }
             }
+        } else {
+            log.debug("[cleanup][未获取到锁，跳过本轮][lockKey={}]", cleanupLockKey);
         }
     }
@@ -59,8 +75,8 @@ public class RedisStreamMessageCleanupJob {
         StreamOperations<String, Object, Object> ops = redisTemplate.getRedisTemplate().opsForStream();
         listeners.forEach(listener -> {
             try {
-                // 使用 XTRIM 命令清理消息，只保留最近的 MAX_LEN 条消息
-                Long trimCount = ops.trim(listener.getStreamKey(), MAX_COUNT, true);
+                // 使用 XTRIM MAXLEN 精确裁剪（approximate=false），避免 ~ 模式下长期明显高于上限
+                Long trimCount = ops.trim(listener.getStreamKey(), MAX_COUNT, false);
                 if (trimCount != null && trimCount > 0) {
                     log.info("[execute][Stream({}) 清理消息数量({})]", listener.getStreamKey(), trimCount);
                 }
@@ -99,7 +99,8 @@ public class IotMessageBusAutoConfiguration {
                                                                             RedisMQTemplate redisTemplate,
                                                                             RedissonClient redissonClient) {
             List<AbstractRedisStreamMessageListener<?>> listeners = getListeners(messageBus);
-            return new RedisPendingMessageResendJob(listeners, redisTemplate, redissonClient);
+            return new RedisPendingMessageResendJob(listeners, redisTemplate, redissonClient,
+                    RedisPendingMessageResendJob.IOT_RESEND_LOCK_KEY);
         }
         /**
@@ -110,7 +111,8 @@ public class IotMessageBusAutoConfiguration {
                                                                             RedisMQTemplate redisTemplate,
                                                                             RedissonClient redissonClient) {
             List<AbstractRedisStreamMessageListener<?>> listeners = getListeners(messageBus);
-            return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient);
+            return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient,
+                    RedisStreamMessageCleanupJob.IOT_CLEANUP_LOCK_KEY);
         }
         private List<AbstractRedisStreamMessageListener<?>> getListeners(IotRedisMessageBus messageBus) {