解决和develop的冲突

ayscb · Jun 1, 2020 · 1a7264d · 1a7264d
2 parents 83e6451 + a8c5a8d
commit 1a7264d
Show file tree

Hide file tree

Showing 107 changed files with 4,055 additions and 1,820 deletions.
diff --git a/client/src/test/java/com/alibaba/nacos/client/ConfigTest.java b/client/src/test/java/com/alibaba/nacos/client/ConfigTest.java
@@ -21,43 +21,64 @@
 import com.alibaba.nacos.api.NacosFactory;
 import com.alibaba.nacos.api.PropertyKeyConst;
 import com.alibaba.nacos.api.config.ConfigService;
+import com.alibaba.nacos.api.config.listener.AbstractListener;
 import com.alibaba.nacos.common.utils.ThreadUtils;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
 
+import java.util.Objects;
 import java.util.Properties;
+import java.util.Scanner;
+import java.util.concurrent.CountDownLatch;
 
 /**
  * @author <a href="mailto:[email protected]">liaochuntao</a>
  */
 @Ignore
 public class ConfigTest {
 
-	private ConfigService configService;
+	private static ConfigService configService;
 
-	@Before
-	public void before() throws Exception {
+	public static void main(String[] args) throws Exception {
+		before();
+		test();
+	}
+
+	public static void before() throws Exception {
 		Properties properties = new Properties();
-		properties.setProperty(PropertyKeyConst.NAMESPACE, "bebf0150-e1ea-47e2-81fe-6814caf2b952");
 		properties.setProperty(PropertyKeyConst.SERVER_ADDR, "127.0.0.1:8848");
-		properties.setProperty(PropertyKeyConst.USERNAME, "chuntaojun");
-		properties.setProperty(PropertyKeyConst.PASSWORD, "1017");
 		configService = NacosFactory.createConfigService(properties);
 	}
 
-	@Test
-	public void test() throws Exception {
+	public static void test() throws Exception {
 		final String dataId = "lessspring";
 		final String group = "lessspring";
 		final String content = "lessspring-" + System.currentTimeMillis();
 		boolean result = configService.publishConfig(dataId, group, content);
 		Assert.assertTrue(result);
 
 		ThreadUtils.sleep(10_000);
-		String response = configService.getConfig(dataId, group, 5000);
-		System.out.println(response);
+
+		String response = configService.getConfigAndSignListener(dataId, group, 5000, new AbstractListener() {
+			@Override
+			public void receiveConfigInfo(String configInfo) {
+				System.err.println(configInfo);
+			}
+		});
+		Assert.assertEquals(content, response);
+
+		Scanner scanner = new Scanner(System.in);
+		System.out.println("input content");
+		while (scanner.hasNextLine()){
+			String s = scanner.next();
+			if (Objects.equals("exit", s)) {
+				scanner.close();
+				return;
+			}
+			configService.publishConfig(dataId, group, s);
+		}
 	}
 
 }
diff --git a/common/pom.xml b/common/pom.xml
@@ -82,8 +82,8 @@
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
                 <configuration>
-                    <source>7</source>
-                    <target>7</target>
+                    <source>6</source>
+                    <target>6</target>
                 </configuration>
             </plugin>
         </plugins>

diff --git a/common/src/main/java/com/alibaba/nacos/common/executor/ExecutorFactory.java b/common/src/main/java/com/alibaba/nacos/common/executor/ExecutorFactory.java
@@ -16,10 +16,8 @@
 
 package com.alibaba.nacos.common.executor;
 
-import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
@@ -39,19 +37,6 @@ public final class ExecutorFactory {
 
     public static final String DEFAULT_NAMESPACE = "nacos";
 
-    public static ForkJoinPool newForkJoinPool(final String group) {
-        ForkJoinPool forkJoinPool = new ForkJoinPool();
-        THREAD_POOL_MANAGER.register(DEFAULT_NAMESPACE, group, forkJoinPool);
-        return forkJoinPool;
-    }
-
-    public static ForkJoinPool newForkJoinPool(final String group,
-                                               final int nThreads) {
-        ForkJoinPool forkJoinPool = new ForkJoinPool(nThreads);
-        THREAD_POOL_MANAGER.register(DEFAULT_NAMESPACE, group, forkJoinPool);
-        return forkJoinPool;
-    }
-
     public static ExecutorService newSingleExecutorService(final String group) {
         ExecutorService executorService = Executors.newFixedThreadPool(1);
         THREAD_POOL_MANAGER.register(DEFAULT_NAMESPACE, group, executorService);

diff --git a/common/src/main/java/com/alibaba/nacos/common/executor/ThreadPoolManager.java b/common/src/main/java/com/alibaba/nacos/common/executor/ThreadPoolManager.java
@@ -17,6 +17,7 @@
 
 
 import com.alibaba.nacos.common.utils.ShutdownUtils;
+import com.alibaba.nacos.common.utils.ThreadUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -69,7 +70,7 @@ public static ThreadPoolManager getInstance() {
     private ThreadPoolManager() {}
 
     private void init() {
-        resourcesManager = new ConcurrentHashMap<>(8);
+        resourcesManager = new ConcurrentHashMap<String, Map<String, Set<ExecutorService>>>(8);
     }
 
     /**
@@ -89,7 +90,7 @@ public void register(String namespace, String group, ExecutorService executor) {
         synchronized (monitor) {
             Map<String, Set<ExecutorService>> map = resourcesManager.get(namespace);
             if (map == null) {
-            	map = new HashMap<>(8);
+            	map = new HashMap<String, Set<ExecutorService>>(8);
             	map.put(group, new HashSet<ExecutorService>());
 				map.get(group).add(executor);
 				resourcesManager.put(namespace, map);
@@ -136,7 +137,12 @@ public void deregister(String namespace, String group, ExecutorService executor)
         }
     }
 
-    public void destroy(String namespace) {
+	/**
+	 * Destroys all thread pool resources under this namespace
+	 *
+	 * @param namespace namespace
+	 */
+	public void destroy(final String namespace) {
 	    final Object monitor = lockers.get(namespace);
 	    if (monitor == null) {
 	        return;
@@ -148,31 +154,36 @@ public void destroy(String namespace) {
 			}
 	        for (Map.Entry<String, Set<ExecutorService>> entry : subResource.entrySet()) {
 	            for (ExecutorService executor : entry.getValue()) {
-					shutdownThreadPool(executor);
+					ThreadUtils.shutdownThreadPool(executor);
 				}
             }
             resourcesManager.get(namespace).clear();
 	        resourcesManager.remove(namespace);
         }
     }
 
-    private void shutdownThreadPool(ExecutorService executor) {
-		executor.shutdown();
-		int retry = 3;
-		while (retry > 0) {
-			retry --;
-			try {
-				if (executor.awaitTermination(10, TimeUnit.SECONDS)) {
-					return;
-				}
-			} catch (InterruptedException e) {
-				executor.shutdownNow();
-				Thread.interrupted();
-			} catch (Throwable ex) {
-				LOGGER.error("ThreadPoolManager shutdown executor has error : {}", ex);
+	/**
+	 * This namespace destroys all thread pool resources under the grouping
+	 *
+	 * @param namespace namespace
+	 * @param group group
+	 */
+	public void destroy(final String namespace, final String group) {
+		final Object monitor = lockers.get(namespace);
+		if (monitor == null) {
+			return;
+		}
+		synchronized (monitor) {
+			Map<String, Set<ExecutorService>> subResource = resourcesManager.get(namespace);
+			if (subResource == null) {
+				return;
+			}
+			Set<ExecutorService> waitDestroy = subResource.get(group);
+			for (ExecutorService executor : waitDestroy) {
+				ThreadUtils.shutdownThreadPool(executor);
 			}
+			resourcesManager.get(namespace).remove(group);
 		}
-		executor.shutdownNow();
 	}
 
     public static void shutdown() {