BJ_ZTEITS-Developers / parking-cloud-yudao

Commit a978c996171017dac9f8ec345f2fc5ab0679004c

Authored by 芋道源码
Committed by Gitee
2 parents 16dd86a8 4faf61a4

!480 解决RedisPendingMessageResendJob 任务可能会导致消息重复的问题 

Merge pull request !480 from 天气真好/redis-stream-mq
Inline Side-by-side
Showing 1 changed file with 29 additions and 16 deletions
yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java
  View file @a978c99
... ... @@ -7,17 +7,14 @@ import lombok.AllArgsConstructor;
7 7 import lombok.extern.slf4j.Slf4j;
8 8 import org.redisson.api.RLock;
9 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 12 import org.springframework.data.redis.core.StreamOperations;
17 13 import org.springframework.scheduling.annotation.Scheduled;
18 14  
19 15 import java.util.List;
20 16 import java.util.Map;
  17 +import java.util.Objects;
21 18  
22 19 /**
23 20 * 这个任务用于处理，crash 之后的消费者未消费完的消息
... ... @@ -32,6 +29,13 @@ public class RedisPendingMessageResendJob {
32 29 private final RedisMQTemplate redisTemplate;
33 30 private final String groupName;
34 31 private final RedissonClient redissonClient;
  32 + /**
  33 + * 消息超时时间，默认5分钟
  34 + * <p>超时的消息才会被重新投递<p/>
  35 + * <p>由于定时任务1分钟一次，消息超时后不会被立即重投，
  36 + * 极端情况下消息5分钟过期后，再等1分钟才会被扫瞄到<p/>
  37 + */
  38 + private final long expireTime = 5 * 60;
35 39  
36 40 /**
37 41 * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题
... ... @@ -54,25 +58,34 @@ public class RedisPendingMessageResendJob {
54 58 private void execute() {
55 59 StreamOperations<String, Object, Object> ops = redisTemplate.getRedisTemplate().opsForStream();
56 60 listeners.forEach(listener -> {
57   - PendingMessagesSummary pendingMessagesSummary = ops.pending(listener.getStreamKey(), groupName);
  61 + PendingMessagesSummary pendingMessagesSummary = Objects.requireNonNull(ops.pending(listener.getStreamKey(), groupName));
58 62 // 每个消费者的 pending 队列消息数量
59 63 Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer();
60 64 pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> {
61 65 log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount);
62   -
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)) {
  66 + // 每个消费者的 pending消息的详情信息
  67 + PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(groupName, consumerName), Range.unbounded(), pendingMessageCount);
  68 + if(pendingMessages.isEmpty()){
66 69 return;
67 70 }
68   - for (MapRecord<String, Object, Object> record : records) {
  71 + for (PendingMessage pendingMessage : pendingMessages) {
  72 + // 获取消息上一次传递到 consumer 的时间,
  73 + long lastDelivery = pendingMessage.getElapsedTimeSinceLastDelivery().getSeconds();
  74 + if(lastDelivery < expireTime){
  75 + continue;
  76 + }
  77 + // 获取指定id的消息体
  78 + List<MapRecord<String, Object, Object>> records = ops.range(listener.getStreamKey(), Range.of(Range.Bound.inclusive(pendingMessage.getIdAsString()), Range.Bound.inclusive(pendingMessage.getIdAsString())));
  79 + if(CollUtil.isEmpty(records)){
  80 + continue;
  81 + }
69 82 // 重新投递消息
70 83 redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord()
71   - .ofObject(record.getValue()) // 设置内容
72   - .withStreamKey(listener.getStreamKey()));
73   -
  84 + .ofObject(records.get(0).getValue()) // 设置内容
  85 + .withStreamKey(listener.getStreamKey()));
74 86 // ack 消息消费完成
75   - redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, record);
  87 + redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, records.get(0));
  88 + log.info("[processPendingMessage][消息({})重新投递成功]", records.get(0).getId());
76 89 }
77 90 });
78 91 });
... ...