refactor: add stream pull nacos config

ozhera-log/log-stream/src/main/java/com/xiaomi/mone/log/stream/config/ConfigManager.java

@@ -77,7 +77,7 @@ public class ConfigManager {
      *
      * @throws StreamException
      */
-    public void initStream() throws StreamException {
+    public void initializeStreamConfig() throws StreamException {
         log.debug("[initStream} nacos dataId:{},group:{}", spaceDataId, DEFAULT_GROUP_ID);
         String streamConfigStr = nacosConfig.getConfigStr(spaceDataId, DEFAULT_GROUP_ID, DEFAULT_TIME_OUT_MS);
         MiLogStreamConfig milogStreamConfig;
@@ -96,19 +96,16 @@ public void initStream() throws StreamException {
                 for (Long spaceId : milogStreamDataMap.keySet()) {
                     final String dataId = milogStreamDataMap.get(spaceId);
                     // init spaceData config
-                    String milogSpaceDataStr = nacosConfig.getConfigStr(dataId, DEFAULT_GROUP_ID, DEFAULT_TIME_OUT_MS);
-                    if (StringUtils.isNotEmpty(milogSpaceDataStr)) {
-                        MilogSpaceData milogSpaceData = GSON.fromJson(milogSpaceDataStr, MilogSpaceData.class);
-                        if (null != milogSpaceData) {
+                    String logSpaceDataStr = nacosConfig.getConfigStr(dataId, DEFAULT_GROUP_ID, DEFAULT_TIME_OUT_MS);
+                    if (StringUtils.isNotEmpty(logSpaceDataStr)) {
+                        MilogSpaceData milogSpaceData = GSON.fromJson(logSpaceDataStr, MilogSpaceData.class);
+                        if (null != milogSpaceData && !milogSpaceDataMap.containsKey(spaceId)) {
+                            MilogConfigListener configListener = new MilogConfigListener(spaceId, dataId, DEFAULT_GROUP_ID, milogSpaceData, nacosConfig);
+                            addListener(spaceId, configListener);
                             milogSpaceDataMap.put(spaceId, milogSpaceData);
+                            log.info("[ConfigManager.initStream] added log config listener for spaceId:{},dataId:{}", spaceId, dataId);
                         }
                     }
-                    MilogSpaceData milogSpaceData = milogSpaceDataMap.get(spaceId);
-                    if (null != milogSpaceData) {
-                        MilogConfigListener configListener = new MilogConfigListener(spaceId, dataId, DEFAULT_GROUP_ID, milogSpaceData, nacosConfig);
-                        addListener(spaceId, configListener);
-                    }
-                    log.info("[ConfigManager.initStream] added log config listener for spaceId:{},dataId:{}", spaceId, dataId);
                 }
             } else {
                 log.info("server start current not contain space config,uniqueMark:{}", uniqueMark);

ozhera-log/log-stream/src/main/java/com/xiaomi/mone/log/stream/job/PullConfigJob.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2020 Xiaomi
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  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.xiaomi.mone.log.stream.job;
+
+import cn.hutool.core.thread.ThreadUtil;
+import com.xiaomi.mone.log.stream.config.ConfigManager;
+import com.xiaomi.youpin.docean.anno.Component;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.annotation.Resource;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ *
+ * @description Regularly pull configuration comparison to compensate for the situation where nacos monitoring does not work.
+ * @version 1.0
+ * @author wtt
+ * @date 2024/4/22 14:08
+ *
+ */
+@Component
+@Slf4j
+public class PullConfigJob {
+
+    @Resource
+    private ConfigManager configManager;
+
+    public void init() {
+
+        log.info("PullConfigJob execute");
+        ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor(
+                ThreadUtil.newNamedThreadFactory("pull-config", false)
+        );
+        long initDelay = 2;
+        long intervalTime = 5;
+        scheduledExecutor.scheduleAtFixedRate(() -> {
+            try {
+                configManager.initializeStreamConfig();
+            } catch (Exception e) {
+                log.error("PullConfigJob execute error", e);
+            }
+        }, initDelay, intervalTime, TimeUnit.MINUTES);
+    }
+}