diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQConsumerAutoConfiguration.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQConsumerAutoConfiguration.java index b802444..5b0d886 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQConsumerAutoConfiguration.java +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQConsumerAutoConfiguration.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> 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> listeners, RedisMQTemplate redisTemplate, RedissonClient redissonClient) { - return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient); + return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient, + RedisStreamMessageCleanupJob.DEFAULT_CLEANUP_LOCK_KEY); } /** diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java index bb16be0..d0b7ae3 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java @@ -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> 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); } } diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisStreamMessageCleanupJob.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisStreamMessageCleanupJob.java index 19da845..eff64fd 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisStreamMessageCleanupJob.java +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisStreamMessageCleanupJob.java @@ -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> 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 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); } diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java index 67ae673..a08b060 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java @@ -99,7 +99,8 @@ public class IotMessageBusAutoConfiguration { RedisMQTemplate redisTemplate, RedissonClient redissonClient) { List> 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> listeners = getListeners(messageBus); - return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient); + return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient, + RedisStreamMessageCleanupJob.IOT_CLEANUP_LOCK_KEY); } private List> getListeners(IotRedisMessageBus messageBus) {