From 0e6b4e463f4e49edea6642c15af7ba87099f2ebc Mon Sep 17 00:00:00 2001 From: gaibu <1016771049@qq.com> Date: Fri, 23 Dec 2022 10:33:38 +0800 Subject: [PATCH] fix: 解决 redis mq 消息丢失问题 --- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java | 2 +- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java | 86 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/scheduler/RedisPendingMessageResendJob.java | 86 -------------------------------------------------------------------------------------- 3 files changed, 87 insertions(+), 87 deletions(-) create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/scheduler/RedisPendingMessageResendJob.java diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java index e306389..042a4f7 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java @@ -8,7 +8,7 @@ import cn.iocoder.yudao.framework.mq.core.RedisMQTemplate; import cn.iocoder.yudao.framework.mq.core.interceptor.RedisMessageInterceptor; import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener; import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessageListener; -import cn.iocoder.yudao.framework.mq.scheduler.RedisPendingMessageResendJob; +import cn.iocoder.yudao.framework.mq.job.RedisPendingMessageResendJob; import cn.iocoder.yudao.framework.redis.config.YudaoRedisAutoConfiguration; import lombok.extern.slf4j.Slf4j; import org.redisson.api.RedissonClient; diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java new file mode 100644 index 0000000..ff01e3e --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java @@ -0,0 +1,86 @@ +package cn.iocoder.yudao.framework.mq.job; + +import cn.hutool.core.collection.CollUtil; +import cn.iocoder.yudao.framework.mq.core.RedisMQTemplate; +import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessageListener; +import lombok.extern.slf4j.Slf4j; +import org.redisson.api.RLock; +import org.redisson.api.RedissonClient; +import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.connection.stream.StreamRecords; +import org.springframework.data.redis.core.StreamOperations; +import org.springframework.scheduling.annotation.Scheduled; + +import java.util.List; +import java.util.Map; + +/** + * 这个任务用于处理，crash 之后的消费者未消费完的消息 + */ +@Slf4j +public class RedisPendingMessageResendJob { + private static final String LOCK_KEY = "redis:pending:msg:lock"; + + private final List> listeners; + private final RedisMQTemplate redisTemplate; + private final String groupName; + private final RedissonClient redissonClient; + + public RedisPendingMessageResendJob(List> listeners, RedisMQTemplate redisTemplate, String groupName, RedissonClient redissonClient) { + this.listeners = listeners; + this.redisTemplate = redisTemplate; + this.groupName = groupName; + this.redissonClient = redissonClient; + } + + /** + * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题 + */ + @Scheduled(cron = "35 * * * * ?") + public void messageResend() { + RLock lock = redissonClient.getLock(LOCK_KEY); + // 尝试加锁 + if (lock.tryLock()) { + try { + execute(); + } catch (Exception ex) { + log.error("[messageResend][执行异常]", ex); + } finally { + lock.unlock(); + } + } + } + + private void execute() { + StreamOperations ops = redisTemplate.getRedisTemplate().opsForStream(); + + listeners.forEach(listener -> { + PendingMessagesSummary pendingMessagesSummary = ops.pending(listener.getStreamKey(), groupName); + // 每个消费者的 pending 队列消息数量 + Map pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer(); + pendingMessagesPerConsumer.entrySet().forEach(entry -> { + String consumerName = entry.getKey(); + Long pendingMessageCount = entry.getValue(); + log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount); + + // 从消费者的 pending 队列中读取消息 + List> records = ops.read(Consumer.from(groupName, consumerName), StreamOffset.create(listener.getStreamKey(), ReadOffset.from("0"))); + if (CollUtil.isNotEmpty(records)) { + for (MapRecord record : records) { + // 重新投递消息 + redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord() + .ofObject(record.getValue()) // 设置内容 + .withStreamKey(listener.getStreamKey())); + + // ack 消息消费完成 + redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, record); + } + } + }); + }); + } +} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/scheduler/RedisPendingMessageResendJob.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/scheduler/RedisPendingMessageResendJob.java deleted file mode 100644 index 2a96805..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/scheduler/RedisPendingMessageResendJob.java +++ /dev/null @@ -1,86 +0,0 @@ -package cn.iocoder.yudao.framework.mq.scheduler; - -import cn.hutool.core.collection.CollUtil; -import cn.iocoder.yudao.framework.mq.core.RedisMQTemplate; -import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessageListener; -import lombok.extern.slf4j.Slf4j; -import org.redisson.api.RLock; -import org.redisson.api.RedissonClient; -import org.springframework.data.redis.connection.stream.Consumer; -import org.springframework.data.redis.connection.stream.MapRecord; -import org.springframework.data.redis.connection.stream.PendingMessagesSummary; -import org.springframework.data.redis.connection.stream.ReadOffset; -import org.springframework.data.redis.connection.stream.StreamOffset; -import org.springframework.data.redis.connection.stream.StreamRecords; -import org.springframework.data.redis.core.StreamOperations; -import org.springframework.scheduling.annotation.Scheduled; - -import java.util.List; -import java.util.Map; - -/** - * 这个任务用于处理，crash 之后的消费者未消费完的消息 - */ -@Slf4j -public class RedisPendingMessageResendJob { - private static final String LOCK_KEY = "redis:pending:msg:lock"; - - private final List> listeners; - private final RedisMQTemplate redisTemplate; - private final String groupName; - private final RedissonClient redissonClient; - - public RedisPendingMessageResendJob(List> listeners, RedisMQTemplate redisTemplate, String groupName, RedissonClient redissonClient) { - this.listeners = listeners; - this.redisTemplate = redisTemplate; - this.groupName = groupName; - this.redissonClient = redissonClient; - } - - /** - * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题 - */ - @Scheduled(cron = "35 * * * * ?") - public void messageResend() { - RLock lock = redissonClient.getLock(LOCK_KEY); - // 尝试加锁 - if (lock.tryLock()) { - try { - execute(); - } catch (Exception ex) { - log.error("[messageResend][执行异常]", ex); - } finally { - lock.unlock(); - } - } - } - - private void execute() { - StreamOperations ops = redisTemplate.getRedisTemplate().opsForStream(); - - listeners.forEach(listener -> { - PendingMessagesSummary pendingMessagesSummary = ops.pending(listener.getStreamKey(), groupName); - // 每个消费者的 pending 队列消息数量 - Map pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer(); - pendingMessagesPerConsumer.entrySet().forEach(entry -> { - String consumerName = entry.getKey(); - Long pendingMessageCount = entry.getValue(); - log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount); - - // 从消费者的 pending 队列中读取消息 - List> records = ops.read(Consumer.from(groupName, consumerName), StreamOffset.create(listener.getStreamKey(), ReadOffset.from("0"))); - if (CollUtil.isNotEmpty(records)) { - for (MapRecord record : records) { - // 重新投递消息 - redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord() - .ofObject(record.getValue()) // 设置内容 - .withStreamKey(listener.getStreamKey())); - - // ack 消息消费完成 - redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, record); - } - } - }); - }); - } -} -- libgit2 0.21.4