From f9451ddac2f4c51ccc30f8b95a09aa933f873c05 Mon Sep 17 00:00:00 2001
From: luofu <luofu@axzo.cn>
Date: Mon, 20 Nov 2023 10:09:05 +0800
Subject: [PATCH] =?UTF-8?q?feat(REQ-1507):=20=E8=80=81=E7=9A=84=E5=BE=85?=
 =?UTF-8?q?=E5=8A=9E=E6=95=B0=E6=8D=AE=E8=BF=81=E7=A7=BB?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

背景：
  https://jira.axzo.cn/browse/REQ-1507?goToView=1

影响：
  无
---
 .../PendingMessageDataInitServiceImpl.java    | 50 ++++++++-----------
 1 file changed, 21 insertions(+), 29 deletions(-)

diff --git a/inside-notices/src/main/java/cn/axzo/msg/center/message/service/impl/PendingMessageDataInitServiceImpl.java b/inside-notices/src/main/java/cn/axzo/msg/center/message/service/impl/PendingMessageDataInitServiceImpl.java
index 6610f765..096335ec 100644
--- a/inside-notices/src/main/java/cn/axzo/msg/center/message/service/impl/PendingMessageDataInitServiceImpl.java
+++ b/inside-notices/src/main/java/cn/axzo/msg/center/message/service/impl/PendingMessageDataInitServiceImpl.java
@@ -31,9 +31,6 @@ import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.UUID;
-import java.util.function.Consumer;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 
 /**
@@ -68,39 +65,35 @@ public class PendingMessageDataInitServiceImpl implements PendingMessageDataInit
                 .ge(MessageRecord::getCreateAt, DateFormatUtil.toDate(queryFrom))
                 .between(MessageRecord::getId, minRecordId, scrollId)
                 .orderByDesc(MessageRecord::getId)
+                .last("LIMIT 1000")
                 .list();
-//        while (!records.isEmpty()) {
-        String profile = environment.getProperty("spring.profiles.active");
-        if ("test".equals(profile) || "master".equals(profile)) {
-            batchSavePendingMessage(records);
-        } else {
-            log.warn("目前只有测试test环境和生产master环境支持该操作！");
+        while (!records.isEmpty()) {
+            String profile = environment.getProperty("spring.profiles.active");
+            if ("test".equals(profile) || "master".equals(profile)) {
+                batchSavePendingMessage(records);
+            } else {
+                log.warn("目前只有测试test环境和生产master环境支持该操作！");
+            }
+            scrollId = records.stream().min(Comparator.comparing(MessageRecord::getId))
+                    .map(MessageRecord::getId).orElse(0L);
+            records = messageRecordDao.lambdaQuery()
+                    .eq(MessageRecord::getType, MsgTypeEnum.PENDING_MESSAGE)
+                    .eq(MessageRecord::getIsDelete, YesNoEnum.NO.getCode())
+                    .ge(MessageRecord::getCreateAt, DateFormatUtil.toDate(queryFrom))
+                    .between(MessageRecord::getId, minRecordId, scrollId)
+                    .orderByDesc(MessageRecord::getId)
+                    .last("LIMIT 1000")
+                    .list();
         }
-//            scrollId = records.stream().min(Comparator.comparing(MessageRecord::getId))
-//                    .map(MessageRecord::getId).orElse(0L);
-//            records = messageRecordDao.lambdaQuery()
-//                    .eq(MessageRecord::getType, MsgTypeEnum.PENDING_MESSAGE)
-//                    .eq(MessageRecord::getIsDelete, YesNoEnum.NO.getCode())
-//                    .ge(MessageRecord::getCreateAt, DateFormatUtil.toDate(queryFrom))
-//                    //.between(MessageRecord::getId, minRecordId, scrollId)
-//                    //modify by zuoqinbo
-//                    .between(MessageRecord::getId, scrollId, Long.MAX_VALUE)
-//                    .orderByDesc(MessageRecord::getId)
-//                    .list();
-//        }
     }
 
     private void batchSavePendingMessage(List<MessageRecord> records) {
-        // TODO: [zuoqinbo]
         if (CollectionUtils.isEmpty(records)) {
             return;
         }
-        List<PendingMessageRecord> pendingMessageRecords = records.stream().map(new Function<MessageRecord, PendingMessageRecord>() {
-            @Override
-            public PendingMessageRecord apply(MessageRecord messageRecord) {
-                return convert(messageRecord);
-            }
-        }).collect(Collectors.toList());
+        List<PendingMessageRecord> pendingMessageRecords = records.stream()
+                .map(this::convert)
+                .collect(Collectors.toList());
         //批量插入数据库
         pendingMessageRecordDao.saveOrUpdateBatch(pendingMessageRecords);
     }
@@ -129,7 +122,6 @@ public class PendingMessageDataInitServiceImpl implements PendingMessageDataInit
         pendingMsgRecord.setOuId(record.getTenantId());
         pendingMsgRecord.setBizCategory(BizCategoryEnum.OTHER);
         pendingMsgRecord.setFailCause(String.valueOf(record.getId()));
-        //
         pendingMsgRecord.setIdentityCode(UUIDUtil.uuidString());
         pendingMsgRecord.setCreateAt(record.getCreateAt());
         pendingMsgRecord.setUpdateAt(record.getUpdateAt());