From c4b56a727c16e643e91ef71289aeb6a8a4e93520 Mon Sep 17 00:00:00 2001
From: caozehui <2427765068@qq.com>
Date: Mon, 11 May 2026 15:49:05 +0800
Subject: [PATCH] =?UTF-8?q?=E5=BE=AE=E8=B0=83?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../module/push/service/message/MessageRecordServiceImpl.java | 2 +-
 .../push/service/retry/MessageRetryQueueServiceImpl.java      | 4 ++++
 2 files changed, 5 insertions(+), 1 deletion(-)

diff --git a/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/message/MessageRecordServiceImpl.java b/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/message/MessageRecordServiceImpl.java
index ff9191a..d507769 100644
--- a/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/message/MessageRecordServiceImpl.java
+++ b/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/message/MessageRecordServiceImpl.java
@@ -276,7 +276,7 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
     public Page<MessageRecordDO> getPage(MessageRecordReqVO reqVO) {
         QueryWrapper<MessageRecordDO> wrapper = new QueryWrapper<>();
         wrapper.lambda()
-                .eq(StrUtil.isNotBlank(reqVO.getMessageType()), MessageRecordDO::getChannel, reqVO.getChannel());
+                .eq(StrUtil.isNotBlank(reqVO.getMessageType()), MessageRecordDO::getMessageType, reqVO.getMessageType());
         return this.page(new Page<>(PageUtils.getPageNum(reqVO), PageUtils.getPageSize(reqVO)), wrapper);
     }
 }
diff --git a/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/retry/MessageRetryQueueServiceImpl.java b/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/retry/MessageRetryQueueServiceImpl.java
index 08de37e..d28363a 100644
--- a/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/retry/MessageRetryQueueServiceImpl.java
+++ b/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/retry/MessageRetryQueueServiceImpl.java
@@ -271,6 +271,7 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
         log.info("========== 重试队列重建完成 ==========");
     }
 
+    // 每60分钟主动从数据库中扫描一次，将消息重试数据同步到 Redis 中
     @Scheduled(fixedRate = 600000)
     @Override
     public void syncRetryQueueConsistency() {
@@ -334,6 +335,9 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
         }
     }
 
+    /**
+     * 通过设置lockUntil时间、process_status=1来从数据库层抢占该条重试消息。
+     */
     private boolean claimRetryMessage(Long messageId, String channel, LocalDateTime currentTime, LocalDateTime lockUntil) {
         return baseMapper.claimRetryMessage(messageId, channel, currentTime, lockUntil) > 0;
     }