fix: 增加加载缓存文件的保护机制，同一个文件最快只能每10s加载一次缓存文件 (#460)

Author: andrewshan

polaris-plugins/polaris-plugins-registry/registry-memory/src/main/java/com/tencent/polaris/plugins/registry/memory/MessagePersistHandler.java

@@ -48,6 +48,11 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.regex.Pattern;
 
@@ -70,6 +75,9 @@ public class MessagePersistHandler {
 
     private static final String PATTERN_SERVICE = "svc#%s#%s#%s.yaml";
 
+    // 同一个文件，两次读取之间的时间间隔为10s
+    private static final long MESSAGE_READ_INTERVAL = TimeUnit.SECONDS.toMillis(10);
+
     private final File persistDirFile;
 
     private final String persistDirPath;
@@ -84,6 +92,9 @@ public class MessagePersistHandler {
 
     private final JsonFormat.Parser parser = JsonFormat.parser();
 
+    // 缓存文件上一次加载的时间，避免频繁对同一个文件进行读取
+    private final Map<ServiceEventKey, Long> messageLastReadTime = new ConcurrentHashMap<>();
+
     public MessagePersistHandler(
             String persistDirPath, int maxWriteRetry, int maxReadRetry, long retryInterval) {
         this.maxReadRetry = maxReadRetry;
@@ -235,6 +246,31 @@ private Path doSaveService(ServiceEventKey svcEventKey, Message message) {
         return persistPath.toAbsolutePath();
     }
 
+    boolean shouldLoadFromStore(ServiceEventKey eventKey) {
+        long currentTimeMs = System.currentTimeMillis();
+        Long previousTimeMs = messageLastReadTime.putIfAbsent(eventKey, currentTimeMs);
+        final boolean[] loadFromStore = {false};
+        if (null != previousTimeMs) {
+            if (currentTimeMs - previousTimeMs < MESSAGE_READ_INTERVAL) {
+                return false;
+            }
+            messageLastReadTime.computeIfPresent(eventKey, new BiFunction<ServiceEventKey, Long, Long>() {
+                @Override
+                public Long apply(ServiceEventKey serviceEventKey, Long aLong) {
+                    if (Objects.equals(aLong, previousTimeMs)) {
+                        loadFromStore[0] = true;
+                        return currentTimeMs;
+                    }
+                    return aLong;
+                }
+            });
+        } else {
+            // first time
+            loadFromStore[0] = true;
+        }
+        return loadFromStore[0];
+    }
+
     /**
      * 遍历缓存目录并加载之前缓存的服务信息
      *
@@ -243,6 +279,9 @@ private Path doSaveService(ServiceEventKey svcEventKey, Message message) {
      * @return 服务标识-消息对象的集合
      */
     public Message loadPersistedServices(ServiceEventKey eventKey, Supplier<Message.Builder> builderSupplier) {
+        if (!shouldLoadFromStore(eventKey)) {
+            return null;
+        }
         String fileName = serviceKeyToFileName(eventKey);
         int retryTimes = 0;
         Message readMessage = null;

polaris-plugins/polaris-plugins-registry/registry-memory/src/test/java/com/tencent/polaris/plugins/registry/memory/MessagePersistHandlerTest.java

@@ -0,0 +1,63 @@
+/*
+ * Tencent is pleased to support the open source community by making Polaris available.
+ *
+ * Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
+ *
+ * Licensed under the BSD 3-Clause License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://opensource.org/licenses/BSD-3-Clause
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed
+ * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+ * CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package com.tencent.polaris.plugins.registry.memory;
+
+import com.tencent.polaris.api.pojo.ServiceEventKey;
+import com.tencent.polaris.api.pojo.ServiceKey;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class MessagePersistHandlerTest {
+
+    @Test
+    public void testMessagePersistHandler_shouldLoadFromStore() {
+        MessagePersistHandler messagePersistHandler = new MessagePersistHandler(
+                "/root", 1, 1, 1000);
+        ServiceEventKey serviceEventKey = new ServiceEventKey(
+                new ServiceKey("Test", "testSvc"), ServiceEventKey.EventType.SERVICE);
+        boolean result1 = messagePersistHandler.shouldLoadFromStore(serviceEventKey);
+        Assert.assertTrue(result1);
+        try {
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+        boolean result2 = messagePersistHandler.shouldLoadFromStore(serviceEventKey);
+        Assert.assertFalse(result2);
+        ServiceEventKey serviceEventKey1 = new ServiceEventKey(
+                new ServiceKey("Test", "testSvc1"), ServiceEventKey.EventType.SERVICE);
+        boolean result11 = messagePersistHandler.shouldLoadFromStore(serviceEventKey1);
+        Assert.assertTrue(result11);
+        try {
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+        boolean result12 = messagePersistHandler.shouldLoadFromStore(serviceEventKey1);
+        Assert.assertFalse(result12);
+        try {
+            Thread.sleep(25000);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+        boolean result3 = messagePersistHandler.shouldLoadFromStore(serviceEventKey);
+        Assert.assertTrue(result3);
+        boolean result13 = messagePersistHandler.shouldLoadFromStore(serviceEventKey1);
+        Assert.assertTrue(result13);
+    }
+}