From 494928ed888a3bbbc14e720b76528a6cb60da3d7 Mon Sep 17 00:00:00 2001 From: wfm Date: Fri, 12 May 2023 17:23:21 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat:=E9=87=8D=E6=94=BE=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E5=88=A4=E6=96=AD=E8=BF=87=E6=9C=9F=E6=97=B6=E9=97=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../mq/job/RedisPendingMessageResendJob.java | 38 ++++++++++--------- 1 file changed, 20 insertions(+), 18 deletions(-) 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 index f4ba050c0..ad0ae833d 100644 --- 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 @@ -7,17 +7,14 @@ import lombok.AllArgsConstructor; 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.domain.Range; +import org.springframework.data.redis.connection.stream.*; import org.springframework.data.redis.core.StreamOperations; import org.springframework.scheduling.annotation.Scheduled; import java.util.List; import java.util.Map; +import java.util.Objects; /** * 这个任务用于处理，crash 之后的消费者未消费完的消息 @@ -33,6 +30,8 @@ public class RedisPendingMessageResendJob { private final String groupName; private final RedissonClient redissonClient; + private final long expireTime = 1000 * 60; + /** * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题 */ @@ -54,25 +53,28 @@ public class RedisPendingMessageResendJob { private void execute() { StreamOperations ops = redisTemplate.getRedisTemplate().opsForStream(); listeners.forEach(listener -> { - PendingMessagesSummary pendingMessagesSummary = ops.pending(listener.getStreamKey(), groupName); + PendingMessagesSummary pendingMessagesSummary = Objects.requireNonNull(ops.pending(listener.getStreamKey(), groupName)); // 每个消费者的 pending 队列消息数量 Map pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer(); pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> { log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount); - // 从消费者的 pending 队列中读取消息 - List> records = ops.read(Consumer.from(groupName, consumerName), StreamOffset.create(listener.getStreamKey(), ReadOffset.from("0"))); - if (CollUtil.isEmpty(records)) { + PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(groupName, consumerName), Range.unbounded(), pendingMessageCount); + if(pendingMessages.isEmpty()){ return; } - for (MapRecord record : records) { - // 重新投递消息 - redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord() - .ofObject(record.getValue()) // 设置内容 - .withStreamKey(listener.getStreamKey())); - - // ack 消息消费完成 - redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, record); + for (PendingMessage pendingMessage : pendingMessages) { + if(pendingMessage.getElapsedTimeSinceLastDelivery().toMillis() - expireTime >= 0){ + List> records = ops.range(listener.getStreamKey(), Range.of(Range.Bound.inclusive(pendingMessage.getIdAsString()), Range.Bound.inclusive(pendingMessage.getIdAsString()))); + if(!CollUtil.isEmpty(records)){ + // 重新投递消息 + redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord() + .ofObject(records.get(0).getValue()) // 设置内容 + .withStreamKey(listener.getStreamKey())); + // ack 消息消费完成 + redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, records.get(0)); + } + } } }); }); From 86be0bdf800209ea79086145b0e26be2bac03bf0 Mon Sep 17 00:00:00 2001 From: wfm Date: Tue, 16 May 2023 09:58:58 +0800 Subject: [PATCH 2/3] fix PR --- .../mq/job/RedisPendingMessageResendJob.java | 36 ++++++++++++------- 1 file changed, 23 insertions(+), 13 deletions(-) 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 index ad0ae833d..4a0fbdfb2 100644 --- 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 @@ -29,8 +29,13 @@ public class RedisPendingMessageResendJob { private final RedisMQTemplate redisTemplate; private final String groupName; private final RedissonClient redissonClient; - - private final long expireTime = 1000 * 60; + /** + * 消息超时时间，默认5分钟 + *

超时的消息才会被重新投递

+ *

由于定时任务1分钟一次，消息超时后不会被立即重投， + * 极端情况下消息5分钟过期后，再等1分钟才会被扫瞄到

+ */ + private final long expireTime = 5 * 60; /** * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题 @@ -58,23 +63,28 @@ public class RedisPendingMessageResendJob { Map pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer(); pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> { log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount); - + // 每个消费者的 pending消息的详情信息 PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(groupName, consumerName), Range.unbounded(), pendingMessageCount); if(pendingMessages.isEmpty()){ return; } for (PendingMessage pendingMessage : pendingMessages) { - if(pendingMessage.getElapsedTimeSinceLastDelivery().toMillis() - expireTime >= 0){ - List> records = ops.range(listener.getStreamKey(), Range.of(Range.Bound.inclusive(pendingMessage.getIdAsString()), Range.Bound.inclusive(pendingMessage.getIdAsString()))); - if(!CollUtil.isEmpty(records)){ - // 重新投递消息 - redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord() - .ofObject(records.get(0).getValue()) // 设置内容 - .withStreamKey(listener.getStreamKey())); - // ack 消息消费完成 - redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, records.get(0)); - } + // 获取消息上一次传递到 consumer 的时间, + long lastDelivery = pendingMessage.getElapsedTimeSinceLastDelivery().getSeconds(); + if(lastDelivery < expireTime){ + continue; } + // 获取指定id的消息体 + List> records = ops.range(listener.getStreamKey(), Range.of(Range.Bound.inclusive(pendingMessage.getIdAsString()), Range.Bound.inclusive(pendingMessage.getIdAsString()))); + if(CollUtil.isEmpty(records)){ + continue; + } + // 重新投递消息 + redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord() + .ofObject(records.get(0).getValue()) // 设置内容 + .withStreamKey(listener.getStreamKey())); + // ack 消息消费完成 + redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, records.get(0)); } }); }); From 4faf61a4750997bff2a08e8dd135de377d99c3d2 Mon Sep 17 00:00:00 2001 From: wfm Date: Tue, 16 May 2023 10:22:46 +0800 Subject: [PATCH 3/3] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../yudao/framework/mq/job/RedisPendingMessageResendJob.java | 1 + 1 file changed, 1 insertion(+) 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 index 4a0fbdfb2..689fd2931 100644 --- 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 @@ -85,6 +85,7 @@ public class RedisPendingMessageResendJob { .withStreamKey(listener.getStreamKey())); // ack 消息消费完成 redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, records.get(0)); + log.info("[processPendingMessage][消息({})重新投递成功]", records.get(0).getId()); } }); });