- add LoadBalancer api

- BinaryWeightRandomLoadBalancer - ArrayWeightRandomLoadBalancer - TreeWeightRandomLoadBalancer - RandomLoadBalancer - add ServiceListenable api - add GovernAutoServiceRegistrationOfNoneWeb
Ahoo-Wang · May 12, 2021 · 87a8a69 · 87a8a69
1 parent ec39aa0
commit 87a8a69
Show file tree

Hide file tree

Showing 38 changed files with 929 additions and 86 deletions.
diff --git a/README.md b/README.md
@@ -13,7 +13,7 @@ process cache refresh, with unparalleled QPS performance and real-time consisten
 ### Gradle
 
 ``` kotlin
-    val governVersion = "0.8.2";
+    val governVersion = "0.9.0";
     implementation("me.ahoo.govern:spring-cloud-starter-config:${governVersion}")
     implementation("me.ahoo.govern:spring-cloud-starter-discovery:${governVersion}")
 ```
@@ -30,7 +30,7 @@ process cache refresh, with unparalleled QPS performance and real-time consisten
   <modelVersion>4.0.0</modelVersion>
   <artifactId>demo</artifactId>
   <properties>
-    <govern.version>0.8.2</govern.version>
+    <govern.version>0.9.0</govern.version>
   </properties>
 
   <dependencies>
@@ -109,14 +109,16 @@ bin/rest-api
 ![rest-api-service](./docs/rest-api-service.png)
 
 - /v1/namespaces/{namespace}/services/
-    - GET
+  - GET
 - /v1/namespaces/{namespace}/services/{serviceId}/instances
-    - GET
-    - PUT
+  - GET
+  - PUT
 - /v1/namespaces/{namespace}/services/{serviceId}/instances/{instanceId}
-    - DELETE
+  - DELETE
 - /v1/namespaces/{namespace}/services/{serviceId}/instances/{instanceId}/metadata
-    - PUT
+  - PUT
+- /v1/namespaces/{namespace}/services/lb/{serviceId}
+  - GET
 
 ## JMH Benchmark
 

diff --git a/README.zh-CN.md b/README.zh-CN.md
@@ -8,7 +8,7 @@ Govern Service* 提供了超高TPS&QPS。*Govern Service* 结合本地进程缓
 ### Gradle
 
 ``` kotlin
-    val governVersion = "0.8.2";
+    val governVersion = "0.9.0";
     implementation("me.ahoo.govern:spring-cloud-starter-config:${governVersion}")
     implementation("me.ahoo.govern:spring-cloud-starter-discovery:${governVersion}")
 ```
@@ -25,7 +25,7 @@ Govern Service* 提供了超高TPS&QPS。*Govern Service* 结合本地进程缓
   <modelVersion>4.0.0</modelVersion>
   <artifactId>demo</artifactId>
   <properties>
-    <govern.version>0.8.2</govern.version>
+    <govern.version>0.9.0</govern.version>
   </properties>
 
   <dependencies>
@@ -112,7 +112,9 @@ bin/rest-api
   - DELETE
 - /v1/namespaces/{namespace}/services/{serviceId}/instances/{instanceId}/metadata
   - PUT
-
+- /v1/namespaces/{namespace}/services/lb/{serviceId}
+  - GET
+
 ## JMH 基准测试
 
 - The development notebook : MacBook Pro (M1)

diff --git a/bom/build.gradle.kts b/bom/build.gradle.kts
@@ -1,9 +1,8 @@
+val libraryProjects = rootProject.ext.get("libraryProjects") as java.util.LinkedHashSet<Project>;
+
 dependencies {
     constraints {
-        rootProject.subprojects.forEach {
-            if (it.name == "rest-api") {
-                return@forEach
-            }
+        libraryProjects.forEach {
             api(it)
         }
     }

diff --git a/build.gradle.kts b/build.gradle.kts
@@ -28,6 +28,7 @@ ext {
     set("jmhVersion", "1.29")
     set("guavaVersion", "30.0-jre")
     set("springfoxVersion", "3.0.0")
+    set("libraryProjects", libraryProjects)
 }
 
 allprojects {

diff --git a/config/src/main/java/me/ahoo/govern/config/ConfigListenable.java b/config/src/main/java/me/ahoo/govern/config/ConfigListenable.java
@@ -7,11 +7,7 @@
  */
 public interface ConfigListenable {
 
-    CompletableFuture<Boolean> addListener(String configId, ConfigChangedListener configChangedListener);
+    CompletableFuture<Boolean> addListener(NamespacedConfigId namespacedConfigId, ConfigChangedListener configChangedListener);
 
-    CompletableFuture<Boolean> addListener(String namespace, String configId, ConfigChangedListener configChangedListener);
-
-    CompletableFuture<Boolean> removeListener(String configId);
-
-    CompletableFuture<Boolean> removeListener(String namespace, String configId);
+    CompletableFuture<Boolean> removeListener(NamespacedConfigId namespacedConfigId, ConfigChangedListener configChangedListener);
 }
diff --git a/config/src/main/java/me/ahoo/govern/config/redis/ConsistencyRedisConfigService.java b/config/src/main/java/me/ahoo/govern/config/redis/ConsistencyRedisConfigService.java
@@ -14,6 +14,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArraySet;
 
 /**
  * @author ahoo wang
@@ -25,7 +26,7 @@ public class ConsistencyRedisConfigService implements ConfigService, ConfigListe
     private final ConfigListener configListener;
 
     private final ConcurrentHashMap<NamespacedConfigId, CompletableFuture<Config>> configMap;
-    private final ConcurrentHashMap<NamespacedConfigId, ConfigChangedListener> configMapListener;
+    private final ConcurrentHashMap<NamespacedConfigId, CopyOnWriteArraySet<ConfigChangedListener>> configMapListener;
 
     public ConsistencyRedisConfigService(ConfigService delegate, MessageListenable messageListenable) {
         this.configMap = new ConcurrentHashMap<>();
@@ -82,26 +83,35 @@ public CompletableFuture<Boolean> removeConfig(String namespace, String configId
         return delegate.removeConfig(namespace, configId);
     }
 
-    @Override
-    public CompletableFuture<Boolean> addListener(String configId, ConfigChangedListener configChangedListener) {
-        return addListener(NamespacedContext.GLOBAL.getNamespace(), configId, configChangedListener);
-    }
 
     @Override
-    public CompletableFuture<Boolean> addListener(String namespace, String configId, ConfigChangedListener configChangedListener) {
-        var putOk = configMapListener.putIfAbsent(NamespacedConfigId.of(namespace, configId), configChangedListener) == null;
-        return CompletableFuture.completedFuture(putOk);
+    public CompletableFuture<Boolean> addListener(NamespacedConfigId namespacedConfigId, ConfigChangedListener configChangedListener) {
+        configMapListener.compute(namespacedConfigId, (key, val) -> {
+            CopyOnWriteArraySet<ConfigChangedListener> listeners = val;
+            if (Objects.isNull(val)) {
+                listeners = new CopyOnWriteArraySet<>();
+            }
+            listeners.add(configChangedListener);
+            return listeners;
+        });
+        return CompletableFuture.completedFuture(true);
     }
 
     @Override
-    public CompletableFuture<Boolean> removeListener(String configId) {
-        return removeListener(NamespacedContext.GLOBAL.getNamespace(), configId);
-    }
+    public CompletableFuture<Boolean> removeListener(NamespacedConfigId namespacedConfigId, ConfigChangedListener configChangedListener) {
+        configMapListener.compute(namespacedConfigId, (key, val) -> {
+            if (Objects.isNull(val)) {
+                return null;
+            }
+            CopyOnWriteArraySet<ConfigChangedListener> listeners = val;
+            listeners.remove(configChangedListener);
+            if (listeners.isEmpty()) {
+                return null;
+            }
+            return listeners;
+        });
 
-    @Override
-    public CompletableFuture<Boolean> removeListener(String namespace, String configId) {
-        var removeOk = configMapListener.remove(NamespacedConfigId.of(namespace, configId)) != null;
-        return CompletableFuture.completedFuture(removeOk);
+        return CompletableFuture.completedFuture(true);
     }
 
     @Override
@@ -146,11 +156,11 @@ public void onMessage(Topic topic, String channel, String message) {
             final var configkey = channel;
             var namespacedConfigId = ConfigKeyGenerator.getConfigIdOfKey(configkey);
             configMap.put(namespacedConfigId, delegate.getConfig(namespacedConfigId.getNamespace(), namespacedConfigId.getConfigId()));
-            var configChangedListener = configMapListener.get(namespacedConfigId);
-            if (Objects.isNull(configChangedListener)) {
+            var configChangedListeners = configMapListener.get(namespacedConfigId);
+            if (Objects.isNull(configChangedListeners) || configChangedListeners.isEmpty()) {
                 return;
             }
-            configChangedListener.onChange(namespacedConfigId, message);
+            configChangedListeners.forEach(configChangedListener -> configChangedListener.onChange(namespacedConfigId, message));
         }
     }
 }
diff --git a/config/src/test/java/me/ahoo/govern/config/redis/ConsistencyRedisConfigServiceTest.java b/config/src/test/java/me/ahoo/govern/config/redis/ConsistencyRedisConfigServiceTest.java
@@ -3,7 +3,7 @@
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import lombok.var;
-import me.ahoo.govern.config.ConfigKeyGenerator;
+import me.ahoo.govern.config.ConfigChangedListener;
 import me.ahoo.govern.config.NamespacedConfigId;
 import me.ahoo.govern.core.listener.RedisMessageListenable;
 import org.junit.jupiter.api.Assertions;
@@ -121,12 +121,16 @@ void getConfigChangedRollback() {
     void addListener() {
         CountDownLatch countDownLatch = new CountDownLatch(1);
         AtomicReference<NamespacedConfigId> changedConfigId = new AtomicReference<>();
-        var testConfig=NamespacedConfigId.of(namespace,testConfigId);
-        var addListenerResult = consistencyRedisConfigService.addListener(namespace, testConfigId, (configId, message) -> {
-            log.warn("addListener@Test - configId:[{}] - message:[{}]", configId, message);
-            changedConfigId.set(configId);
-            countDownLatch.countDown();
-        });
+        var testConfig = NamespacedConfigId.of(namespace, testConfigId);
+        var changedListener = new ConfigChangedListener() {
+            @Override
+            public void onChange(NamespacedConfigId namespacedConfigId, String op) {
+                log.warn("addListener@Test - configId:[{}] - message:[{}]", namespacedConfigId, op);
+                changedConfigId.set(namespacedConfigId);
+                countDownLatch.countDown();
+            }
+        };
+        var addListenerResult = consistencyRedisConfigService.addListener(testConfig, changedListener);
         Assertions.assertTrue(addListenerResult.join());
         getConfigChanged();
         countDownLatch.await();
@@ -135,11 +139,17 @@ void addListener() {
 
     @Test
     void removeListener() {
-        var addListenerResult = consistencyRedisConfigService.addListener(namespace, testConfigId, (configId, message) -> {
-            Assertions.fail();
-        });
+        var testConfig = NamespacedConfigId.of(namespace, testConfigId);
+        var changedListener = new ConfigChangedListener() {
+            @Override
+            public void onChange(NamespacedConfigId namespacedConfigId, String op) {
+                Assertions.fail();
+            }
+        };
+        var addListenerResult = consistencyRedisConfigService.addListener(testConfig, changedListener);
+
         Assertions.assertTrue(addListenerResult.join());
-        var removeListenerResult = consistencyRedisConfigService.removeListener(namespace, testConfigId);
+        var removeListenerResult = consistencyRedisConfigService.removeListener(testConfig, changedListener);
         Assertions.assertTrue(removeListenerResult.join());
         getConfigChanged();
     }

diff --git a/core/src/main/java/me/ahoo/govern/core/util/RedisScripts.java b/core/src/main/java/me/ahoo/govern/core/util/RedisScripts.java
@@ -14,10 +14,13 @@
 /**
  * @author ahoo wang
  */
-public class RedisScripts {
+public final class RedisScripts {
     private static final String WARN_CLEAR_TEST_DATA = "warn_clear_test_data.lua";
     public static final ConcurrentHashMap<String, CompletableFuture<String>> scriptMapSha = new ConcurrentHashMap<>();
 
+    private RedisScripts() {
+    }
+
     @SneakyThrows
     public static String getScript(String scriptName) {
         URL url = Resources.getResource(scriptName);
@@ -31,6 +34,13 @@ public static CompletableFuture<String> loadScript(String scriptName, RedisScrip
         });
     }
 
+    public static CompletableFuture<String> reloadScript(String scriptName, RedisScriptingAsyncCommands<String, String> scriptingCommands) {
+        return scriptMapSha.compute(scriptName, (key, result) -> {
+            String script = getScript(key);
+            return scriptingCommands.scriptLoad(script).toCompletableFuture();
+        });
+    }
+
 
     /**
      * only for dev

diff --git a/core/src/main/java/me/ahoo/govern/core/util/Systems.java b/core/src/main/java/me/ahoo/govern/core/util/Systems.java
@@ -0,0 +1,22 @@
+package me.ahoo.govern.core.util;
+
+import java.lang.management.ManagementFactory;
+
+/**
+ * @author ahoo wang
+ */
+public final class Systems {
+    private Systems() {
+    }
+
+
+    public static String getCurrentProcessName() {
+        return ManagementFactory.getRuntimeMXBean().getName();
+    }
+
+    public static long getCurrentProcessId() {
+        String processName = getCurrentProcessName();
+        String processIdStr = processName.split("@")[0];
+        return Long.parseLong(processIdStr);
+    }
+}
diff --git a/core/src/test/java/me/ahoo/govern/core/util/SystemsTest.java b/core/src/test/java/me/ahoo/govern/core/util/SystemsTest.java
@@ -0,0 +1,23 @@
+package me.ahoo.govern.core.util;
+
+import lombok.var;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author ahoo wang
+ */
+class SystemsTest {
+
+    @Test
+    public void getCurrentProcessName() {
+        var currentProcessName = Systems.getCurrentProcessName();
+        Assertions.assertNotNull(currentProcessName);
+    }
+
+    @Test
+    public void getCurrentProcessId() {
+        var processId = Systems.getCurrentProcessId();
+        Assertions.assertTrue(processId > 0);
+    }
+}
diff --git a/discovery/src/main/java/me/ahoo/govern/discovery/RenewProperties.java b/discovery/src/main/java/me/ahoo/govern/discovery/RenewProperties.java
@@ -1,16 +1,27 @@
 package me.ahoo.govern.discovery;
 
-import lombok.Getter;
-import lombok.Setter;
-
 /**
  * @author ahoo wang
  */
 public class RenewProperties {
-    @Getter
-    @Setter
+
     private int initialDelay = 1;
-    @Getter
-    @Setter
+
     private int period = 10;
+
+    public int getInitialDelay() {
+        return initialDelay;
+    }
+
+    public void setInitialDelay(int initialDelay) {
+        this.initialDelay = initialDelay;
+    }
+
+    public int getPeriod() {
+        return period;
+    }
+
+    public void setPeriod(int period) {
+        this.period = period;
+    }
 }
diff --git a/...oo/govern/discovery/ServiceEventType.java → ...ern/discovery/ServiceChangedListener.java b/...oo/govern/discovery/ServiceEventType.java → ...ern/discovery/ServiceChangedListener.java
@@ -3,10 +3,12 @@
 /**
  * @author ahoo wang
  */
-public interface ServiceEventType {
+public interface ServiceChangedListener {
     String REGISTER = "register";
     String DEREGISTER = "deregister";
     String EXPIRED = "expired";
     String RENEW = "renew";
     String SET_METADATA = "set_metadata";
+
+    void onChange(NamespacedServiceId namespacedServiceId, String op);
 }
diff --git a/discovery/src/main/java/me/ahoo/govern/discovery/ServiceInstance.java b/discovery/src/main/java/me/ahoo/govern/discovery/ServiceInstance.java
@@ -10,6 +10,7 @@
  * @author ahoo wang
  */
 public class ServiceInstance extends Instance {
+
     public static final ServiceInstance NOT_FOUND = new ServiceInstance();
     private int weight = 1;
     private boolean ephemeral = true;

diff --git a/discovery/src/main/java/me/ahoo/govern/discovery/ServiceListenable.java b/discovery/src/main/java/me/ahoo/govern/discovery/ServiceListenable.java
@@ -0,0 +1,11 @@
+package me.ahoo.govern.discovery;
+
+/**
+ * @author ahoo wang
+ */
+public interface ServiceListenable {
+
+    void addListener(NamespacedServiceId namespacedServiceId, ServiceChangedListener serviceChangedListener);
+
+    void removeListener(NamespacedServiceId namespacedServiceId, ServiceChangedListener serviceChangedListener);
+}