From aca65eb1f5e1f7103dc08e89b1ecdb35becb2dcc Mon Sep 17 00:00:00 2001
From: opensnail <598092184@qq.com>
Date: Tue, 16 Jul 2024 18:21:46 +0800
Subject: [PATCH] =?UTF-8?q?fix(sj=5F1.1.1):=20=E4=BC=98=E5=8C=96=E5=8E=BB?=
 =?UTF-8?q?=E9=87=8D=E5=A4=84=E7=90=86?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../server/common/register/AbstractRegister.java     | 11 ++++++++++-
 .../server/common/register/ClientRegister.java       | 12 +++++-------
 2 files changed, 15 insertions(+), 8 deletions(-)

diff --git a/snail-job-server/snail-job-server-common/src/main/java/com/aizuda/snailjob/server/common/register/AbstractRegister.java b/snail-job-server/snail-job-server-common/src/main/java/com/aizuda/snailjob/server/common/register/AbstractRegister.java
index dbeded3ea..8c9f899fb 100644
--- a/snail-job-server/snail-job-server-common/src/main/java/com/aizuda/snailjob/server/common/register/AbstractRegister.java
+++ b/snail-job-server/snail-job-server-common/src/main/java/com/aizuda/snailjob/server/common/register/AbstractRegister.java
@@ -74,12 +74,21 @@ public abstract class AbstractRegister implements Register, Lifecycle {
             .map(serverNode -> Pair.of(serverNode.getHostId(), serverNode.getHostIp())).collect(
                 Collectors.toSet());
 
+        // 去重处理
+        Set<Pair<String, String>> existed = Sets.newHashSet();
         for (final ServerNode serverNode : serverNodes) {
-            if (pairs.contains(Pair.of(serverNode.getHostId(), serverNode.getHostIp()))) {
+            Pair<String, String> pair = Pair.of(serverNode.getHostId(), serverNode.getHostIp());
+            if (existed.contains(pair)) {
+                continue;
+            }
+
+            if (pairs.contains(pair)) {
                 updateDBs.add(serverNode);
             } else {
                 insertDBs.add(serverNode);
             }
+
+            existed.add(pair);
         }
 
         try {
diff --git a/snail-job-server/snail-job-server-common/src/main/java/com/aizuda/snailjob/server/common/register/ClientRegister.java b/snail-job-server/snail-job-server-common/src/main/java/com/aizuda/snailjob/server/common/register/ClientRegister.java
index 94f8ca1b3..29eee112d 100644
--- a/snail-job-server/snail-job-server-common/src/main/java/com/aizuda/snailjob/server/common/register/ClientRegister.java
+++ b/snail-job-server/snail-job-server-common/src/main/java/com/aizuda/snailjob/server/common/register/ClientRegister.java
@@ -3,6 +3,7 @@ package com.aizuda.snailjob.server.common.register;
 import com.aizuda.snailjob.common.core.constant.SystemConstants.HTTP_PATH;
 import com.aizuda.snailjob.common.core.enums.NodeTypeEnum;
 import com.aizuda.snailjob.common.log.SnailJobLog;
+import com.aizuda.snailjob.server.common.triple.Pair;
 import com.aizuda.snailjob.template.datasource.persistence.po.ServerNode;
 import com.google.common.collect.Lists;
 import lombok.extern.slf4j.Slf4j;
@@ -12,6 +13,8 @@ import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -87,14 +90,9 @@ public class ClientRegister extends AbstractRegister implements Runnable {
                 ServerNode serverNode = QUEUE.poll(5L, TimeUnit.SECONDS);
                 if (Objects.nonNull(serverNode)) {
                     List<ServerNode> lists = Lists.newArrayList(serverNode);
-                    QUEUE.drainTo(lists, 100);
-
-                    // 去重
-                    lists = new ArrayList<>(lists.stream()
-                            .collect(
-                                    Collectors.toMap(ServerNode::getHostId, node -> node, (existing, replacement) -> existing))
-                            .values());
+                    QUEUE.drainTo(lists, 256);
 
+                    // 注册或续租
                     refreshExpireAt(lists);
                 }
             } catch (InterruptedException ignored) {