BJ_ZTEITS-Developers / parking-cloud-yudao

Commit e9645d7054db531e02910c7ea89f35b09aea2fe4

Authored by gaibu
1 parent 46f37904

fix: 解决 redis mq 消息丢失问题

Inline Side-by-side
Showing 1 changed file with 5 additions and 5 deletions
yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/scheduler/RedisPendingMessageResendJob.java
  View file @e9645d7
@@ -61,17 +61,17 @@ public class RedisPendingMessageResendJob { @@ -61,17 +61,17 @@ public class RedisPendingMessageResendJob {
61 61
62 listeners.forEach(listener -> { 62 listeners.forEach(listener -> {
63 PendingMessagesSummary pendingMessagesSummary = ops.pending(listener.getStreamKey(), groupName); 63 PendingMessagesSummary pendingMessagesSummary = ops.pending(listener.getStreamKey(), groupName);
64 - // 每个消费者的pending消息数量 64 + // 每个消费者的 pending 队列消息数量
65 Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer(); 65 Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer();
66 pendingMessagesPerConsumer.entrySet().forEach(entry -> { 66 pendingMessagesPerConsumer.entrySet().forEach(entry -> {
67 String consumerName = entry.getKey(); 67 String consumerName = entry.getKey();
68 Long pendingMessageCount = entry.getValue(); 68 Long pendingMessageCount = entry.getValue();
69 log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount); 69 log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount);
70 70
71 - // 从消费者的pending队列中读取消息  
72 - List<MapRecord<String, Object, Object>> retVal = ops.read(Consumer.from(groupName, consumerName), StreamOffset.create(listener.getStreamKey(), ReadOffset.from("0")));  
73 - if (CollUtil.isNotEmpty(retVal)) {  
74 - for (MapRecord<String, Object, Object> record : retVal) { 71 + // 从消费者的 pending 队列中读取消息
  72 + List<MapRecord<String, Object, Object>> records = ops.read(Consumer.from(groupName, consumerName), StreamOffset.create(listener.getStreamKey(), ReadOffset.from("0")));
  73 + if (CollUtil.isNotEmpty(records)) {
  74 + for (MapRecord<String, Object, Object> record : records) {
75 // 重新投递消息 75 // 重新投递消息
76 redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord() 76 redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord()
77 .ofObject(record.getValue()) // 设置内容 77 .ofObject(record.getValue()) // 设置内容