BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 86be0bdf800209ea79086145b0e26be2bac03bf0

Authored by wfm
1 parent 494928ed

fix PR

Inline Side-by-side
Showing 1 changed file with 23 additions and 13 deletions
yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java
  View file @86be0bd
@@ -29,8 +29,13 @@ public class RedisPendingMessageResendJob { @@ -29,8 +29,13 @@ public class RedisPendingMessageResendJob {
29 private final RedisMQTemplate redisTemplate; 29 private final RedisMQTemplate redisTemplate;
30 private final String groupName; 30 private final String groupName;
31 private final RedissonClient redissonClient; 31 private final RedissonClient redissonClient;
32 -  
33 - private final long expireTime = 1000 * 60; 32 + /**
  33 + * 消息超时时间，默认5分钟
  34 + * <p>超时的消息才会被重新投递<p/>
  35 + * <p>由于定时任务1分钟一次，消息超时后不会被立即重投，
  36 + * 极端情况下消息5分钟过期后，再等1分钟才会被扫瞄到<p/>
  37 + */
  38 + private final long expireTime = 5 * 60;
34 39
35 /** 40 /**
36 * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题 41 * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题
@@ -58,23 +63,28 @@ public class RedisPendingMessageResendJob { @@ -58,23 +63,28 @@ public class RedisPendingMessageResendJob {
58 Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer(); 63 Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer();
59 pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> { 64 pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> {
60 log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount); 65 log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount);
61 - 66 + // 每个消费者的 pending消息的详情信息
62 PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(groupName, consumerName), Range.unbounded(), pendingMessageCount); 67 PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(groupName, consumerName), Range.unbounded(), pendingMessageCount);
63 if(pendingMessages.isEmpty()){ 68 if(pendingMessages.isEmpty()){
64 return; 69 return;
65 } 70 }
66 for (PendingMessage pendingMessage : pendingMessages) { 71 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 - } 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;
77 } 81 }
  82 + // 重新投递消息
  83 + redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord()
  84 + .ofObject(records.get(0).getValue()) // 设置内容
  85 + .withStreamKey(listener.getStreamKey()));
  86 + // ack 消息消费完成
  87 + redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, records.get(0));
78 } 88 }
79 }); 89 });
80 }); 90 });