BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 494928ed888a3bbbc14e720b76528a6cb60da3d7

Authored by wfm
1 parent 711a2763

feat:重放消息判断过期时间

Inline Side-by-side
Showing 1 changed file with 20 additions and 18 deletions
yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java
  View file @494928e
@@ -7,17 +7,14 @@ import lombok.AllArgsConstructor; @@ -7,17 +7,14 @@ import lombok.AllArgsConstructor;
7 import lombok.extern.slf4j.Slf4j; 7 import lombok.extern.slf4j.Slf4j;
8 import org.redisson.api.RLock; 8 import org.redisson.api.RLock;
9 import org.redisson.api.RedissonClient; 9 import org.redisson.api.RedissonClient;
10 -import org.springframework.data.redis.connection.stream.Consumer;  
11 -import org.springframework.data.redis.connection.stream.MapRecord;  
12 -import org.springframework.data.redis.connection.stream.PendingMessagesSummary;  
13 -import org.springframework.data.redis.connection.stream.ReadOffset;  
14 -import org.springframework.data.redis.connection.stream.StreamOffset;  
15 -import org.springframework.data.redis.connection.stream.StreamRecords; 10 +import org.springframework.data.domain.Range;
  11 +import org.springframework.data.redis.connection.stream.*;
16 import org.springframework.data.redis.core.StreamOperations; 12 import org.springframework.data.redis.core.StreamOperations;
17 import org.springframework.scheduling.annotation.Scheduled; 13 import org.springframework.scheduling.annotation.Scheduled;
18 14
19 import java.util.List; 15 import java.util.List;
20 import java.util.Map; 16 import java.util.Map;
  17 +import java.util.Objects;
21 18
22 /** 19 /**
23 * 这个任务用于处理，crash 之后的消费者未消费完的消息 20 * 这个任务用于处理，crash 之后的消费者未消费完的消息
@@ -33,6 +30,8 @@ public class RedisPendingMessageResendJob { @@ -33,6 +30,8 @@ public class RedisPendingMessageResendJob {
33 private final String groupName; 30 private final String groupName;
34 private final RedissonClient redissonClient; 31 private final RedissonClient redissonClient;
35 32
  33 + private final long expireTime = 1000 * 60;
  34 +
36 /** 35 /**
37 * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题 36 * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题
38 */ 37 */
@@ -54,25 +53,28 @@ public class RedisPendingMessageResendJob { @@ -54,25 +53,28 @@ public class RedisPendingMessageResendJob {
54 private void execute() { 53 private void execute() {
55 StreamOperations<String, Object, Object> ops = redisTemplate.getRedisTemplate().opsForStream(); 54 StreamOperations<String, Object, Object> ops = redisTemplate.getRedisTemplate().opsForStream();
56 listeners.forEach(listener -> { 55 listeners.forEach(listener -> {
57 - PendingMessagesSummary pendingMessagesSummary = ops.pending(listener.getStreamKey(), groupName); 56 + PendingMessagesSummary pendingMessagesSummary = Objects.requireNonNull(ops.pending(listener.getStreamKey(), groupName));
58 // 每个消费者的 pending 队列消息数量 57 // 每个消费者的 pending 队列消息数量
59 Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer(); 58 Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer();
60 pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> { 59 pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> {
61 log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount); 60 log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount);
62 61
63 - // 从消费者的 pending 队列中读取消息  
64 - List<MapRecord<String, Object, Object>> records = ops.read(Consumer.from(groupName, consumerName), StreamOffset.create(listener.getStreamKey(), ReadOffset.from("0")));  
65 - if (CollUtil.isEmpty(records)) { 62 + PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(groupName, consumerName), Range.unbounded(), pendingMessageCount);
  63 + if(pendingMessages.isEmpty()){
66 return; 64 return;
67 } 65 }
68 - for (MapRecord<String, Object, Object> record : records) {  
69 - // 重新投递消息  
70 - redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord()  
71 - .ofObject(record.getValue()) // 设置内容  
72 - .withStreamKey(listener.getStreamKey()));  
73 -  
74 - // ack 消息消费完成  
75 - redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, record); 66 + for (PendingMessage pendingMessage : pendingMessages) {
  67 + if(pendingMessage.getElapsedTimeSinceLastDelivery().toMillis() - expireTime >= 0){
  68 + List<MapRecord<String, Object, Object>> records = ops.range(listener.getStreamKey(), Range.of(Range.Bound.inclusive(pendingMessage.getIdAsString()), Range.Bound.inclusive(pendingMessage.getIdAsString())));
  69 + if(!CollUtil.isEmpty(records)){
  70 + // 重新投递消息
  71 + redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord()
  72 + .ofObject(records.get(0).getValue()) // 设置内容
  73 + .withStreamKey(listener.getStreamKey()));
  74 + // ack 消息消费完成
  75 + redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, records.get(0));
  76 + }
  77 + }
76 } 78 }
77 }); 79 });
78 }); 80 });