fix #141 ExecutorEngine parameter error

apache · Sep 1, 2016 · c4077e0 · c4077e0
1 parent 577ae0b
commit c4077e0
Show file tree

Hide file tree

Showing 23 changed files with 280 additions and 39 deletions.
diff --git a/...re/src/main/java/com/dangdang/ddframe/rdb/sharding/config/ShardingPropertiesConstant.java b/...re/src/main/java/com/dangdang/ddframe/rdb/sharding/config/ShardingPropertiesConstant.java
@@ -70,10 +70,10 @@ public enum ShardingPropertiesConstant {
      * 最大工作线程数量.
      * 
      * <p>
-     * 默认值: CPU的核数 * 2
+     * 默认值: 100
      * </p>
      */
-    EXECUTOR_MAX_SIZE("executor.max.size", String.valueOf(Runtime.getRuntime().availableProcessors() * 2), int.class),
+    EXECUTOR_MAX_SIZE("executor.max.size", "100", int.class),
 
     /**
      * 工作线程空闲时超时时间.

diff --git a/...ng-jdbc-core/src/main/java/com/dangdang/ddframe/rdb/sharding/executor/ExecutorEngine.java b/...ng-jdbc-core/src/main/java/com/dangdang/ddframe/rdb/sharding/executor/ExecutorEngine.java
@@ -19,6 +19,7 @@
 
 import com.dangdang.ddframe.rdb.sharding.config.ShardingProperties;
 import com.dangdang.ddframe.rdb.sharding.config.ShardingPropertiesConstant;
+import com.dangdang.ddframe.rdb.sharding.exception.ShardingJdbcException;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -32,7 +33,7 @@
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
@@ -51,7 +52,7 @@ public ExecutorEngine(final ShardingProperties shardingProperties) {
         int executorMaxSize = shardingProperties.getValue(ShardingPropertiesConstant.EXECUTOR_MAX_SIZE);
         long executorMaxIdleTimeoutMilliseconds = shardingProperties.getValue(ShardingPropertiesConstant.EXECUTOR_MAX_IDLE_TIMEOUT_MILLISECONDS);
         executorService = MoreExecutors.listeningDecorator(MoreExecutors.getExitingExecutorService(
-                new ThreadPoolExecutor(executorMinIdleSize, executorMaxSize, executorMaxIdleTimeoutMilliseconds, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())));
+                new ThreadPoolExecutor(executorMinIdleSize, executorMaxSize, executorMaxIdleTimeoutMilliseconds, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>())));
     }
 
     /**
@@ -84,6 +85,20 @@ public <I, M, O> O execute(final Collection<I> inputs, final ExecuteUnit<I, M> e
         return mergeUnit.merge(execute(inputs, executeUnit));
     }
 
+    /**
+     * 安全关闭执行器,并释放线程.
+     */
+    public void shutdown() {
+        executorService.shutdownNow();
+        try {
+            executorService.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (final InterruptedException ignored) {
+        }
+        if (!executorService.isTerminated()) {
+            throw new ShardingJdbcException("ExecutorEngine can not been terminated");
+        }
+    }
+
     private <I, O> ListenableFuture<List<O>> submitFutures(final Collection<I> inputs, final ExecuteUnit<I, O> executeUnit) {
         Set<ListenableFuture<O>> result = new HashSet<>(inputs.size());
         for (final I each : inputs) {

diff --git a/...ng-jdbc-core/src/main/java/com/dangdang/ddframe/rdb/sharding/jdbc/ShardingDataSource.java b/...ng-jdbc-core/src/main/java/com/dangdang/ddframe/rdb/sharding/jdbc/ShardingDataSource.java
@@ -41,6 +41,8 @@ public class ShardingDataSource extends AbstractDataSourceAdapter {
 
     private final ShardingProperties shardingProperties;
 
+    private final ExecutorEngine executorEngine;
+
     private final ShardingContext shardingContext;
 
     public ShardingDataSource(final ShardingRule shardingRule) {
@@ -51,8 +53,9 @@ public ShardingDataSource(final ShardingRule shardingRule, final Properties prop
         Preconditions.checkNotNull(shardingRule);
         Preconditions.checkNotNull(props);
         shardingProperties = new ShardingProperties(props);
+        executorEngine = new ExecutorEngine(shardingProperties);
         try {
-            shardingContext = new ShardingContext(shardingRule, new SQLRouteEngine(shardingRule, DatabaseType.valueFrom(getDatabaseProductName(shardingRule))), new ExecutorEngine(shardingProperties));
+            shardingContext = new ShardingContext(shardingRule, new SQLRouteEngine(shardingRule, DatabaseType.valueFrom(getDatabaseProductName(shardingRule))), executorEngine);
         } catch (final SQLException ex) {
             throw new ShardingJdbcException(ex);
         }
@@ -80,4 +83,11 @@ public ShardingConnection getConnection() throws SQLException {
         MetricsContext.init(shardingProperties);
         return new ShardingConnection(shardingContext);
     }
+
+    /**
+     * 关闭数据源,释放相关资源.
+     */
+    public void shutdown() {
+        executorEngine.shutdown();
+    }
 }
diff --git a/...t/java/com/dangdang/ddframe/rdb/integrate/db/AbstractShardingDataBasesOnlyDBUnitTest.java b/...t/java/com/dangdang/ddframe/rdb/integrate/db/AbstractShardingDataBasesOnlyDBUnitTest.java
@@ -27,14 +27,17 @@
 import com.dangdang.ddframe.rdb.sharding.api.strategy.table.NoneTableShardingAlgorithm;
 import com.dangdang.ddframe.rdb.sharding.api.strategy.table.TableShardingStrategy;
 import com.dangdang.ddframe.rdb.sharding.jdbc.ShardingDataSource;
+import org.junit.AfterClass;
 
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
 public abstract class AbstractShardingDataBasesOnlyDBUnitTest extends AbstractDBUnitTest {
 
-    private final String dataSourceName = "dataSource_%s";
+    private static boolean isShutdown;
+
+    private static ShardingDataSource shardingDataSource;
 
     @Override
     protected List<String> getSchemaFiles() {
@@ -67,13 +70,24 @@ protected List<String> getDataSetFiles() {
     }
 
     protected final ShardingDataSource getShardingDataSource() {
-        DataSourceRule dataSourceRule = new DataSourceRule(createDataSourceMap(dataSourceName));
+        if (null != shardingDataSource && !isShutdown) {
+            return shardingDataSource;
+        }
+        isShutdown = false;
+        DataSourceRule dataSourceRule = new DataSourceRule(createDataSourceMap("dataSource_%s"));
         TableRule orderTableRule = TableRule.builder("t_order").dataSourceRule(dataSourceRule).build();
         TableRule orderItemTableRule = TableRule.builder("t_order_item").dataSourceRule(dataSourceRule).build();
         ShardingRule shardingRule = ShardingRule.builder().dataSourceRule(dataSourceRule).tableRules(Arrays.asList(orderTableRule, orderItemTableRule))
                 .bindingTableRules(Collections.singletonList(new BindingTableRule(Arrays.asList(orderTableRule, orderItemTableRule))))
                 .databaseShardingStrategy(new DatabaseShardingStrategy(Collections.singletonList("user_id"), new MultipleKeysModuloDatabaseShardingAlgorithm()))
                 .tableShardingStrategy(new TableShardingStrategy(Collections.singletonList("order_id"), new NoneTableShardingAlgorithm())).build();
-        return new ShardingDataSource(shardingRule);
+        shardingDataSource = new ShardingDataSource(shardingRule);
+        return shardingDataSource;
+    }
+
+    @AfterClass
+    public static void clear() {
+        isShutdown = true;
+        shardingDataSource.shutdown();
     }
 }
diff --git a/...e/rdb/integrate/dbtbl/dynamic/pstatement/DynamicShardingBothForPStatementWithDMLTest.java b/...e/rdb/integrate/dbtbl/dynamic/pstatement/DynamicShardingBothForPStatementWithDMLTest.java
@@ -21,6 +21,7 @@
 import com.dangdang.ddframe.rdb.integrate.dbtbl.dynamic.DynamicShardingBothHelper;
 import com.dangdang.ddframe.rdb.sharding.jdbc.ShardingDataSource;
 import org.dbunit.DatabaseUnitException;
+import org.junit.AfterClass;
 import org.junit.Test;
 
 import java.sql.Connection;
@@ -29,9 +30,20 @@
 
 public final class DynamicShardingBothForPStatementWithDMLTest extends AbstractShardingBothForPStatementWithDMLTest {
 
+    private static ShardingDataSource shardingDataSource;
+
     @Override
     protected ShardingDataSource getShardingDataSource() {
-        return DynamicShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        if (null != shardingDataSource) {
+            return shardingDataSource;
+        }
+        shardingDataSource = DynamicShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        return shardingDataSource;
+    }
+
+    @AfterClass
+    public static void clear() {
+        shardingDataSource.shutdown();
     }
 
     @Test(expected = IllegalStateException.class)

diff --git a/...db/integrate/dbtbl/dynamic/pstatement/DynamicShardingBothForPStatementWithSelectTest.java b/...db/integrate/dbtbl/dynamic/pstatement/DynamicShardingBothForPStatementWithSelectTest.java
@@ -21,15 +21,27 @@
 import com.dangdang.ddframe.rdb.integrate.dbtbl.dynamic.DynamicShardingBothHelper;
 import com.dangdang.ddframe.rdb.sharding.jdbc.ShardingDataSource;
 import org.dbunit.DatabaseUnitException;
+import org.junit.AfterClass;
 import org.junit.Test;
 
 import java.sql.SQLException;
 
 public final class DynamicShardingBothForPStatementWithSelectTest extends AbstractShardingBothForPStatementWithSelectTest {
 
+    private static ShardingDataSource shardingDataSource;
+
     @Override
     protected ShardingDataSource getShardingDataSource() {
-        return DynamicShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        if (null != shardingDataSource) {
+            return shardingDataSource;
+        }
+        shardingDataSource = DynamicShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        return shardingDataSource;
+    }
+
+    @AfterClass
+    public static void clear() {
+        shardingDataSource.shutdown();
     }
 
     @Test(expected = UnsupportedOperationException.class)

diff --git a/...b/integrate/dbtbl/dynamic/statement/DynamicShardingBothForStatementWithAggregateTest.java b/...b/integrate/dbtbl/dynamic/statement/DynamicShardingBothForStatementWithAggregateTest.java
@@ -21,18 +21,27 @@
 import com.dangdang.ddframe.rdb.integrate.dbtbl.dynamic.DynamicShardingBothHelper;
 import com.dangdang.ddframe.rdb.sharding.jdbc.ShardingDataSource;
 import org.dbunit.DatabaseUnitException;
+import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.Test;
 
 import java.sql.SQLException;
 
 public final class DynamicShardingBothForStatementWithAggregateTest extends AbstractShardingBothTest {
 
-    private ShardingDataSource shardingDataSource;
+    private static ShardingDataSource shardingDataSource;
 
     @Before
-    public void init() throws SQLException {
-        shardingDataSource = getShardingDataSource();
+    public void init() {
+        if (null != shardingDataSource) {
+            return;
+        }
+        shardingDataSource = DynamicShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+    }
+
+    @AfterClass
+    public static void clear() {
+        shardingDataSource.shutdown();
     }
 
     @Override

diff --git a/...ame/rdb/integrate/dbtbl/dynamic/statement/DynamicShardingBothForStatementWithDMLTest.java b/...ame/rdb/integrate/dbtbl/dynamic/statement/DynamicShardingBothForStatementWithDMLTest.java
@@ -21,6 +21,7 @@
 import com.dangdang.ddframe.rdb.integrate.dbtbl.dynamic.DynamicShardingBothHelper;
 import com.dangdang.ddframe.rdb.sharding.jdbc.ShardingDataSource;
 import org.dbunit.DatabaseUnitException;
+import org.junit.AfterClass;
 import org.junit.Test;
 
 import java.sql.Connection;
@@ -29,9 +30,20 @@
 
 public final class DynamicShardingBothForStatementWithDMLTest extends AbstractShardingBothForStatementWithDMLTest {
 
+    private static ShardingDataSource shardingDataSource;
+
     @Override
     protected ShardingDataSource getShardingDataSource() {
-        return DynamicShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        if (null != shardingDataSource) {
+            return shardingDataSource;
+        }
+        shardingDataSource = DynamicShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        return shardingDataSource;
+    }
+
+    @AfterClass
+    public static void clear() {
+        shardingDataSource.shutdown();
     }
 
     @Test(expected = IllegalStateException.class)

diff --git a/.../rdb/integrate/dbtbl/dynamic/statement/DynamicShardingBothForStatementWithSelectTest.java b/.../rdb/integrate/dbtbl/dynamic/statement/DynamicShardingBothForStatementWithSelectTest.java
@@ -21,15 +21,27 @@
 import com.dangdang.ddframe.rdb.integrate.dbtbl.dynamic.DynamicShardingBothHelper;
 import com.dangdang.ddframe.rdb.sharding.jdbc.ShardingDataSource;
 import org.dbunit.DatabaseUnitException;
+import org.junit.AfterClass;
 import org.junit.Test;
 
 import java.sql.SQLException;
 
 public final class DynamicShardingBothForStatementWithSelectTest extends AbstractShardingBothForStatementWithSelectTest {
 
+    private static ShardingDataSource shardingDataSource;
+
     @Override
     protected ShardingDataSource getShardingDataSource() {
-        return DynamicShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        if (null != shardingDataSource) {
+            return shardingDataSource;
+        }
+        shardingDataSource = DynamicShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        return shardingDataSource;
+    }
+
+    @AfterClass
+    public static void clear() {
+        shardingDataSource.shutdown();
     }
 
     @Test(expected = UnsupportedOperationException.class)

diff --git a/...tegrate/dbtbl/statically/pstatement/StaticShardingBothForPStatementWithAggregateTest.java b/...tegrate/dbtbl/statically/pstatement/StaticShardingBothForPStatementWithAggregateTest.java
@@ -20,11 +20,22 @@
 import com.dangdang.ddframe.rdb.integrate.dbtbl.common.pstatement.AbstractShardingBothForPStatementWithAggregateTest;
 import com.dangdang.ddframe.rdb.integrate.dbtbl.statically.StaticShardingBothHelper;
 import com.dangdang.ddframe.rdb.sharding.jdbc.ShardingDataSource;
+import org.junit.AfterClass;
 
 public final class StaticShardingBothForPStatementWithAggregateTest extends AbstractShardingBothForPStatementWithAggregateTest {
+    private static ShardingDataSource shardingDataSource;
 
     @Override
     protected ShardingDataSource getShardingDataSource() {
-        return StaticShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        if (null != shardingDataSource) {
+            return shardingDataSource;
+        }
+        shardingDataSource = StaticShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        return shardingDataSource;
+    }
+
+    @AfterClass
+    public static void clear() {
+        shardingDataSource.shutdown();
     }
 }
diff --git a/...rdb/integrate/dbtbl/statically/pstatement/StaticShardingBothForPStatementWithDMLTest.java b/...rdb/integrate/dbtbl/statically/pstatement/StaticShardingBothForPStatementWithDMLTest.java
@@ -21,6 +21,7 @@
 import com.dangdang.ddframe.rdb.integrate.dbtbl.statically.StaticShardingBothHelper;
 import com.dangdang.ddframe.rdb.sharding.jdbc.ShardingDataSource;
 import org.dbunit.DatabaseUnitException;
+import org.junit.AfterClass;
 import org.junit.Test;
 
 import java.sql.Connection;
@@ -32,9 +33,20 @@
 
 public final class StaticShardingBothForPStatementWithDMLTest extends AbstractShardingBothForPStatementWithDMLTest {
 
+    private static ShardingDataSource shardingDataSource;
+
     @Override
     protected ShardingDataSource getShardingDataSource() {
-        return StaticShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        if (null != shardingDataSource) {
+            return shardingDataSource;
+        }
+        shardingDataSource = StaticShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        return shardingDataSource;
+    }
+
+    @AfterClass
+    public static void clear() {
+        shardingDataSource.shutdown();
     }
 
     @Test

diff --git a/...integrate/dbtbl/statically/pstatement/StaticShardingBothForPStatementWithGroupByTest.java b/...integrate/dbtbl/statically/pstatement/StaticShardingBothForPStatementWithGroupByTest.java
@@ -20,11 +20,23 @@
 import com.dangdang.ddframe.rdb.integrate.dbtbl.common.pstatement.AbstractShardingBothForPStatementWithGroupByTest;
 import com.dangdang.ddframe.rdb.integrate.dbtbl.statically.StaticShardingBothHelper;
 import com.dangdang.ddframe.rdb.sharding.jdbc.ShardingDataSource;
+import org.junit.AfterClass;
 
 public final class StaticShardingBothForPStatementWithGroupByTest extends AbstractShardingBothForPStatementWithGroupByTest {
 
+    private static ShardingDataSource shardingDataSource;
+
     @Override
     protected ShardingDataSource getShardingDataSource() {
-        return StaticShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        if (null != shardingDataSource) {
+            return shardingDataSource;
+        }
+        shardingDataSource = StaticShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        return shardingDataSource;
+    }
+
+    @AfterClass
+    public static void clear() {
+        shardingDataSource.shutdown();
     }
 }
diff --git a/.../integrate/dbtbl/statically/pstatement/StaticShardingBothForPStatementWithSelectTest.java b/.../integrate/dbtbl/statically/pstatement/StaticShardingBothForPStatementWithSelectTest.java
@@ -21,15 +21,27 @@
 import com.dangdang.ddframe.rdb.integrate.dbtbl.statically.StaticShardingBothHelper;
 import com.dangdang.ddframe.rdb.sharding.jdbc.ShardingDataSource;
 import org.dbunit.DatabaseUnitException;
+import org.junit.AfterClass;
 import org.junit.Test;
 
 import java.sql.SQLException;
 
 public final class StaticShardingBothForPStatementWithSelectTest extends AbstractShardingBothForPStatementWithSelectTest {
 
+    private static ShardingDataSource shardingDataSource;
+
     @Override
     protected ShardingDataSource getShardingDataSource() {
-        return StaticShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        if (null != shardingDataSource) {
+            return shardingDataSource;
+        }
+        shardingDataSource = StaticShardingBothHelper.getShardingDataSource(createDataSourceMap("dataSource_%s"));
+        return shardingDataSource;
+    }
+
+    @AfterClass
+    public static void clear() {
+        shardingDataSource.shutdown();
     }
 
     @Test