server cluster synchro 核心逻辑异步化

Author: saleson

spring-cloud-gray-server-cluster/src/main/java/cn/springcloud/gray/server/clustering/synchro/SimpleSynchDataAcceptor.java

@@ -7,6 +7,10 @@
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author saleson
@@ -17,8 +21,16 @@ public class SimpleSynchDataAcceptor implements SynchDataAcceptor {
 
     private List<SynchDataListener> synchDataListeners;
     private Map<String, List<SynchDataListener>> dataTypeListenerCache = new HashMap<>();
+    private ExecutorService executorService;
 
     public SimpleSynchDataAcceptor(List<SynchDataListener> synchDataListeners) {
+        this(new ThreadPoolExecutor(10, 10,
+                0L, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<Runnable>(10)), synchDataListeners);
+    }
+
+    public SimpleSynchDataAcceptor(ExecutorService executorService, List<SynchDataListener> synchDataListeners) {
+        this.executorService = executorService;
         this.synchDataListeners = synchDataListeners;
         refreshDataTypeListenerCache();
     }
@@ -31,10 +43,11 @@ public void accept(SynchData synchData) {
         if (Objects.isNull(listeners)) {
             return;
         }
-        //todo 计划改为异步
-        for (SynchDataListener synchDataListener : listeners) {
-            synchDataListener.listen(synchData);
-        }
+        executorService.execute(() -> {
+            for (SynchDataListener synchDataListener : listeners) {
+                synchDataListener.listen(synchData);
+            }
+        });
     }
 
     private void refreshDataTypeListenerCache() {

spring-cloud-gray-server-plugins/spring-cloud-gray-server-cluster-http-synchro-plugin/src/main/java/cn/springcloud/gray/server/clustering/synchro/http/HttpServerSynchronizer.java

@@ -13,6 +13,10 @@
 import java.io.IOException;
 import java.io.ObjectOutputStream;
 import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author saleson
@@ -24,20 +28,27 @@ public class HttpServerSynchronizer implements ServerSynchronizer {
     private int peerNodeSynchronizeRetryTimes;
     private RestTemplate rest;
     private ServerCluster serverCluster;
-
+    private ExecutorService executorService;
 
     public HttpServerSynchronizer(ServerCluster serverCluster) {
-        this(new RestTemplate(), serverCluster);
+        this(serverCluster, new ThreadPoolExecutor(10, 10,
+                0L, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<Runnable>(10)));
+    }
+
+    public HttpServerSynchronizer(ServerCluster serverCluster, ExecutorService executorService) {
+        this(new RestTemplate(), serverCluster, executorService);
     }
 
-    public HttpServerSynchronizer(RestTemplate rest, ServerCluster serverCluster) {
-        this(3, rest, serverCluster);
+    public HttpServerSynchronizer(RestTemplate rest, ServerCluster serverCluster, ExecutorService executorService) {
+        this(3, rest, serverCluster, executorService);
     }
 
-    public HttpServerSynchronizer(int peerNodeSynchronizeRetryTimes, RestTemplate rest, ServerCluster serverCluster) {
+    public HttpServerSynchronizer(int peerNodeSynchronizeRetryTimes, RestTemplate rest, ServerCluster serverCluster, ExecutorService executorService) {
         this.peerNodeSynchronizeRetryTimes = peerNodeSynchronizeRetryTimes;
         this.rest = rest;
         this.serverCluster = serverCluster;
+        this.executorService = executorService;
     }
 
     @Override
@@ -46,13 +57,14 @@ public void broadcast(SynchData synchData) {
         if (ArrayUtils.isEmpty(peerNodes)) {
             return;
         }
-        //todo 计划改为异步
-        if (StringUtils.isEmpty(synchData.getId())) {
-            synchData.setId(StringUtils.defaultString(synchData.getDataType()) +
-                    cn.springcloud.gray.utils.StringUtils.generateUUID());
-        }
+        executorService.execute(() -> {
+            if (StringUtils.isEmpty(synchData.getId())) {
+                synchData.setId(StringUtils.defaultString(synchData.getDataType()) +
+                        cn.springcloud.gray.utils.StringUtils.generateUUID());
+            }
 
-        broadcastSynchData(synchData, peerNodes);
+            broadcastSynchData(synchData, peerNodes);
+        });
     }
 
     private void broadcastSynchData(SynchData synchData, String[] peerNodes) {

spring-cloud-gray-server-plugins/spring-cloud-gray-server-cluster-http-synchro-plugin/src/main/java/cn/springcloud/gray/server/clustering/synchro/http/configuration/HttpSynchroAutoConfiguration.java

@@ -4,10 +4,13 @@
 import cn.springcloud.gray.server.clustering.synchro.ServerSynchronizer;
 import cn.springcloud.gray.server.clustering.synchro.http.HttpServerSynchronizer;
 import cn.springcloud.gray.server.clustering.synchro.http.ServerSynchDataAcceptEndpoint;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import java.util.concurrent.ExecutorService;
+
 /**
  * @author saleson
  * @date 2020-08-16 04:48
@@ -17,8 +20,10 @@ public class HttpSynchroAutoConfiguration {
 
     @Bean
     @ConditionalOnMissingBean
-    public ServerSynchronizer serverSynchronizer(ServerCluster serverCluster) {
-        return new HttpServerSynchronizer(serverCluster);
+    public ServerSynchronizer serverSynchronizer(
+            @Qualifier("synchroExecutorService") ExecutorService synchroExecutorService,
+            ServerCluster serverCluster) {
+        return new HttpServerSynchronizer(serverCluster, synchroExecutorService);
     }
 
     @Bean

spring-cloud-gray-server/src/main/java/cn/springcloud/gray/server/configuration/SynchroAutoConfiguration.java

@@ -1,15 +1,23 @@
 package cn.springcloud.gray.server.configuration;
 
+import cn.springcloud.gray.concurrent.ExecutorConcurrentStrategy;
 import cn.springcloud.gray.server.clustering.synchro.GrayEventSynchroListener;
 import cn.springcloud.gray.server.clustering.synchro.SimpleSynchDataAcceptor;
 import cn.springcloud.gray.server.clustering.synchro.SynchDataAcceptor;
 import cn.springcloud.gray.server.clustering.synchro.SynchDataListener;
+import cn.springcloud.gray.server.configuration.properties.ClusterProperties;
 import cn.springlcoud.gray.event.server.GrayEventSender;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author saleson
@@ -18,11 +26,27 @@
 @Configuration
 public class SynchroAutoConfiguration {
 
+    @Autowired
+    private ClusterProperties clusterProperties;
+
+
+    @Bean
+    @ConditionalOnMissingBean(name = "synchroExecutorService")
+    public ExecutorService synchroExecutorService() {
+        ExecutorConcurrentStrategy ecs = clusterProperties.
+                getSynchro().getExecutorConcurrentStrategy();
+        return new ThreadPoolExecutor(ecs.getCorePoolSize(), ecs.getMaximumPoolSize(),
+                0L, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<Runnable>(ecs.getQueueSize()));
+    }
+
 
     @Bean
     @ConditionalOnMissingBean
-    public SynchDataAcceptor synchDataAcceptor(List<SynchDataListener> synchDataListeners) {
-        return new SimpleSynchDataAcceptor(synchDataListeners);
+    public SynchDataAcceptor synchDataAcceptor(
+            @Qualifier("synchroExecutorService") ExecutorService synchroExecutorService,
+            List<SynchDataListener> synchDataListeners) {
+        return new SimpleSynchDataAcceptor(synchroExecutorService, synchDataListeners);
     }
 
 

spring-cloud-gray-server/src/main/java/cn/springcloud/gray/server/configuration/properties/ClusterProperties.java

@@ -1,5 +1,6 @@
 package cn.springcloud.gray.server.configuration.properties;
 
+import cn.springcloud.gray.concurrent.ExecutorConcurrentStrategy;
 import cn.springcloud.gray.server.clustering.PeerNode;
 import lombok.Data;
 import org.springframework.boot.context.properties.ConfigurationProperties;
@@ -15,4 +16,12 @@
 @ConfigurationProperties(prefix = "gray.server.cluster")
 public class ClusterProperties {
     private List<PeerNode> peerNodes = new ArrayList<>();
+    private SynchroProperties synchro = new SynchroProperties();
+
+    @Data
+    public class SynchroProperties {
+        private ExecutorConcurrentStrategy executorConcurrentStrategy = new ExecutorConcurrentStrategy();
+
+    }
+
 }

spring-cloud-gray-utils/src/main/java/cn/springcloud/gray/concurrent/ExecutorConcurrentStrategy.java

@@ -0,0 +1,20 @@
+package cn.springcloud.gray.concurrent;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * @author saleson
+ * @date 2020-08-16 16:41
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+public class ExecutorConcurrentStrategy {
+    private int corePoolSize = 10;
+    private int maximumPoolSize = 10;
+    private int queueSize = 10;
+}