for #1172, add ExecutorDataMap to ShardingExecuteEngine

apache · Sep 25, 2018 · 2d67276 · 2d67276
1 parent 7028929
commit 2d67276
Show file tree

Hide file tree

Showing 7 changed files with 27 additions and 39 deletions.
diff --git a/sharding-core/src/main/java/io/shardingsphere/core/executor/ShardingExecuteEngine.java b/sharding-core/src/main/java/io/shardingsphere/core/executor/ShardingExecuteEngine.java
@@ -22,6 +22,7 @@
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import io.shardingsphere.core.exception.ShardingException;
+import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
 
 import java.sql.SQLException;
 import java.util.ArrayList;
@@ -30,6 +31,7 @@
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -90,11 +92,13 @@ public <I, O> List<O> execute(final Collection<I> inputs, final ShardingExecuteC
 
     private <I, O> Collection<ListenableFuture<O>> asyncExecute(final Collection<I> inputs, final ShardingExecuteCallback<I, O> callback) {
         Collection<ListenableFuture<O>> result = new ArrayList<>(inputs.size());
+        final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
         for (final I each : inputs) {
             result.add(executorService.submit(new Callable<O>() {
 
                 @Override
                 public O call() throws SQLException {
+                    ExecutorDataMap.setDataMap(dataMap);
                     return callback.execute(each, false);
                 }
             }));
@@ -164,10 +168,12 @@ private <I, O> Collection<ListenableFuture<Collection<O>>> asyncGroupExecute(fin
     }
 
     private <I, O> ListenableFuture<Collection<O>> asyncGroupExecute(final ShardingExecuteGroup<I> inputGroup, final ShardingGroupExecuteCallback<I, O> callback) {
+        final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
         return executorService.submit(new Callable<Collection<O>>() {
 
             @Override
             public Collection<O> call() throws SQLException {
+                ExecutorDataMap.setDataMap(dataMap);
                 return callback.execute(inputGroup.getInputs(), false);
             }
         });

diff --git a/...ng-core/src/main/java/io/shardingsphere/core/executor/sql/execute/SQLExecuteCallback.java b/...ng-core/src/main/java/io/shardingsphere/core/executor/sql/execute/SQLExecuteCallback.java
@@ -26,7 +26,6 @@
 import io.shardingsphere.core.executor.ShardingExecuteCallback;
 import io.shardingsphere.core.executor.ShardingGroupExecuteCallback;
 import io.shardingsphere.core.executor.StatementExecuteUnit;
-import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
 import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
 import io.shardingsphere.core.metadata.datasource.DataSourceMetaData;
 import io.shardingsphere.core.metadata.datasource.DataSourceMetaDataFactory;
@@ -38,7 +37,6 @@
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
 
 /**
  * Statement execute callback interface.
@@ -57,12 +55,8 @@ public abstract class SQLExecuteCallback<T> implements ShardingExecuteCallback<S
 
     private final boolean isExceptionThrown;
 
-    private final Map<String, Object> dataMap;
-
     private final EventBus shardingEventBus = ShardingEventBusInstance.getInstance();
 
-    private final SQLExecutionHook sqlExecutionHook = new SPISQLExecutionHook();
-
     @Override
     public final T execute(final StatementExecuteUnit statementExecuteUnit, final boolean isTrunkThread) throws SQLException {
         return execute0(statementExecuteUnit, isTrunkThread);
@@ -79,9 +73,9 @@ public final Collection<T> execute(final Collection<StatementExecuteUnit> statem
 
     private T execute0(final StatementExecuteUnit statementExecuteUnit, final boolean isTrunkThread) throws SQLException {
         ExecutorExceptionHandler.setExceptionThrown(isExceptionThrown);
-        ExecutorDataMap.setDataMap(dataMap);
         List<List<Object>> parameterSets = statementExecuteUnit.getRouteUnit().getSqlUnit().getParameterSets();
         DataSourceMetaData dataSourceMetaData = DataSourceMetaDataFactory.newInstance(databaseType, statementExecuteUnit.getStatement().getConnection().getMetaData().getURL());
+        SQLExecutionHook sqlExecutionHook = new SPISQLExecutionHook();
         for (List<Object> each : parameterSets) {
             sqlExecutionHook.start(statementExecuteUnit.getRouteUnit().getDataSourceName(), statementExecuteUnit.getRouteUnit().getSqlUnit().getSql(), each, dataSourceMetaData, isTrunkThread);
             // TODO remove after BED removed

diff --git a/...ng-jdbc/src/main/java/io/shardingsphere/core/executor/BatchPreparedStatementExecutor.java b/...ng-jdbc/src/main/java/io/shardingsphere/core/executor/BatchPreparedStatementExecutor.java
@@ -25,7 +25,6 @@
 import io.shardingsphere.core.constant.ConnectionMode;
 import io.shardingsphere.core.constant.DatabaseType;
 import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
-import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
 import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
 import io.shardingsphere.core.executor.sql.prepare.SQLExecutePrepareCallback;
 import io.shardingsphere.core.jdbc.core.connection.ShardingConnection;
@@ -42,7 +41,6 @@
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
 import java.util.Map.Entry;
 
 /**
@@ -158,8 +156,7 @@ private void handleNewRouteUnits(final Collection<BatchRouteUnit> newRouteUnits)
      */
     public int[] executeBatch() throws SQLException {
         final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
-        final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
-        SQLExecuteCallback<int[]> callback = new SQLExecuteCallback<int[]>(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
+        SQLExecuteCallback<int[]> callback = new SQLExecuteCallback<int[]>(getDatabaseType(), getSqlType(), isExceptionThrown) {
 
             @Override
             protected int[] executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {

diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/PreparedStatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/PreparedStatementExecutor.java
@@ -21,7 +21,6 @@
 import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
 import io.shardingsphere.core.executor.sql.execute.result.MemoryQueryResult;
 import io.shardingsphere.core.executor.sql.execute.result.StreamQueryResult;
-import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
 import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
 import io.shardingsphere.core.executor.sql.prepare.SQLExecutePrepareCallback;
 import io.shardingsphere.core.jdbc.core.connection.ShardingConnection;
@@ -37,7 +36,6 @@
 import java.sql.Statement;
 import java.util.Collection;
 import java.util.List;
-import java.util.Map;
 
 /**
  * Prepared statement executor.
@@ -99,8 +97,7 @@ private PreparedStatement createPreparedStatement(final Connection connection, f
      */
     public List<QueryResult> executeQuery() throws SQLException {
         final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
-        final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
-        SQLExecuteCallback<QueryResult> executeCallback = new SQLExecuteCallback<QueryResult>(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
+        SQLExecuteCallback<QueryResult> executeCallback = new SQLExecuteCallback<QueryResult>(getDatabaseType(), getSqlType(), isExceptionThrown) {
 
             @Override
             protected QueryResult executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
@@ -125,8 +122,7 @@ private QueryResult getQueryResult(final StatementExecuteUnit statementExecuteUn
      */
     public int executeUpdate() throws SQLException {
         final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
-        final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
-        SQLExecuteCallback<Integer> executeCallback = new SQLExecuteCallback<Integer>(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
+        SQLExecuteCallback<Integer> executeCallback = new SQLExecuteCallback<Integer>(getDatabaseType(), getSqlType(), isExceptionThrown) {
 
             @Override
             protected Integer executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
@@ -153,8 +149,7 @@ private int accumulate(final List<Integer> results) {
      */
     public boolean execute() throws SQLException {
         boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
-        Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
-        SQLExecuteCallback<Boolean> executeCallback = new SQLExecuteCallback<Boolean>(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
+        SQLExecuteCallback<Boolean> executeCallback = new SQLExecuteCallback<Boolean>(getDatabaseType(), getSqlType(), isExceptionThrown) {
 
             @Override
             protected Boolean executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {

diff --git a/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/StatementExecutor.java b/sharding-jdbc/src/main/java/io/shardingsphere/core/executor/StatementExecutor.java
@@ -21,7 +21,6 @@
 import io.shardingsphere.core.executor.sql.execute.SQLExecuteCallback;
 import io.shardingsphere.core.executor.sql.execute.result.MemoryQueryResult;
 import io.shardingsphere.core.executor.sql.execute.result.StreamQueryResult;
-import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
 import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
 import io.shardingsphere.core.executor.sql.prepare.SQLExecutePrepareCallback;
 import io.shardingsphere.core.jdbc.core.connection.ShardingConnection;
@@ -35,7 +34,6 @@
 import java.sql.Statement;
 import java.util.Collection;
 import java.util.List;
-import java.util.Map;
 
 /**
  * Statement executor.
@@ -88,8 +86,7 @@ public StatementExecuteUnit createStatementExecuteUnit(final Connection connecti
      */
     public List<QueryResult> executeQuery() throws SQLException {
         final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
-        final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
-        SQLExecuteCallback<QueryResult> executeCallback = new SQLExecuteCallback<QueryResult>(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
+        SQLExecuteCallback<QueryResult> executeCallback = new SQLExecuteCallback<QueryResult>(getDatabaseType(), getSqlType(), isExceptionThrown) {
 
             @Override
             protected QueryResult executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
@@ -174,8 +171,7 @@ public int executeUpdate(final Statement statement, final String sql) throws SQL
 
     private int executeUpdate(final Updater updater) throws SQLException {
         final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
-        final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
-        SQLExecuteCallback<Integer> executeCallback = new SQLExecuteCallback<Integer>(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
+        SQLExecuteCallback<Integer> executeCallback = new SQLExecuteCallback<Integer>(getDatabaseType(), getSqlType(), isExceptionThrown) {
 
             @Override
             protected Integer executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {
@@ -263,8 +259,7 @@ public boolean execute(final Statement statement, final String sql) throws SQLEx
 
     private boolean execute(final Executor executor) throws SQLException {
         final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
-        final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
-        SQLExecuteCallback<Boolean> executeCallback = new SQLExecuteCallback<Boolean>(getDatabaseType(), getSqlType(), isExceptionThrown, dataMap) {
+        SQLExecuteCallback<Boolean> executeCallback = new SQLExecuteCallback<Boolean>(getDatabaseType(), getSqlType(), isExceptionThrown) {
 
             @Override
             protected Boolean executeSQL(final StatementExecuteUnit statementExecuteUnit) throws SQLException {

diff --git a/...tracing/src/main/java/io/shardingsphere/opentracing/hook/OpenTracingSQLExecutionHook.java b/...tracing/src/main/java/io/shardingsphere/opentracing/hook/OpenTracingSQLExecutionHook.java
@@ -61,16 +61,20 @@ public void start(final String dataSourceName, final String sql, final List<Obje
 
     @Override
     public void finishSuccess() {
-        span.finish();
+        if (null != span) {
+            span.finish();
+        }
         if (null != activeSpan) {
             activeSpan.deactivate();
         }
     }
 
     @Override
     public void finishFailure(final Exception cause) {
-        ShardingErrorSpan.setError(span, cause);
-        span.finish();
+        if (null != span) {
+            ShardingErrorSpan.setError(span, cause);
+            span.finish();
+        }
         if (null != activeSpan) {
             activeSpan.deactivate();
         }

diff --git a/...g-proxy/src/main/java/io/shardingsphere/proxy/backend/jdbc/execute/JDBCExecuteEngine.java b/...g-proxy/src/main/java/io/shardingsphere/proxy/backend/jdbc/execute/JDBCExecuteEngine.java
@@ -28,7 +28,6 @@
 import io.shardingsphere.core.executor.sql.execute.SQLExecuteTemplate;
 import io.shardingsphere.core.executor.sql.execute.result.MemoryQueryResult;
 import io.shardingsphere.core.executor.sql.execute.result.StreamQueryResult;
-import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorDataMap;
 import io.shardingsphere.core.executor.sql.execute.threadlocal.ExecutorExceptionHandler;
 import io.shardingsphere.core.executor.sql.prepare.SQLExecutePrepareCallback;
 import io.shardingsphere.core.executor.sql.prepare.SQLExecutePrepareTemplate;
@@ -65,7 +64,6 @@
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
 
 /**
  * SQL Execute engine for JDBC.
@@ -111,12 +109,11 @@ public ExecuteResponse execute(final SQLRouteResult routeResult) throws SQLExcep
         boolean isReturnGeneratedKeys = routeResult.getSqlStatement() instanceof InsertStatement;
         SQLType sqlType = routeResult.getSqlStatement().getType();
         boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
-        Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
         Collection<ShardingExecuteGroup<StatementExecuteUnit>> sqlExecuteGroups =
                 sqlExecutePrepareTemplate.getExecuteUnitGroups(routeResult.getRouteUnits(), new ProxyJDBCExecutePrepareCallback(isReturnGeneratedKeys));
         Collection<ExecuteResponseUnit> executeResponseUnits = sqlExecuteTemplate.executeGroup((Collection) sqlExecuteGroups,
-                new FirstProxyJDBCExecuteCallback(sqlType, isExceptionThrown, dataMap, isReturnGeneratedKeys),
-                new ProxyJDBCExecuteCallback(sqlType, isExceptionThrown, dataMap, isReturnGeneratedKeys));
+                new FirstProxyJDBCExecuteCallback(sqlType, isExceptionThrown, isReturnGeneratedKeys),
+                new ProxyJDBCExecuteCallback(sqlType, isExceptionThrown, isReturnGeneratedKeys));
         ExecuteResponseUnit firstExecuteResponseUnit = executeResponseUnits.iterator().next();
         return firstExecuteResponseUnit instanceof ExecuteQueryResponseUnit
                 ? getExecuteQueryResponse(((ExecuteQueryResponseUnit) firstExecuteResponseUnit).getQueryResponsePackets(), executeResponseUnits) : new ExecuteUpdateResponse(executeResponseUnits);
@@ -200,8 +197,8 @@ private final class FirstProxyJDBCExecuteCallback extends SQLExecuteCallback<Exe
 
         private boolean hasMetaData;
 
-        private FirstProxyJDBCExecuteCallback(final SQLType sqlType, final boolean isExceptionThrown, final Map<String, Object> dataMap, final boolean isReturnGeneratedKeys) {
-            super(DatabaseType.MySQL, sqlType, isExceptionThrown, dataMap);
+        private FirstProxyJDBCExecuteCallback(final SQLType sqlType, final boolean isExceptionThrown, final boolean isReturnGeneratedKeys) {
+            super(DatabaseType.MySQL, sqlType, isExceptionThrown);
             this.isReturnGeneratedKeys = isReturnGeneratedKeys;
         }
 
@@ -222,8 +219,8 @@ private final class ProxyJDBCExecuteCallback extends SQLExecuteCallback<ExecuteR
 
         private final boolean isReturnGeneratedKeys;
 
-        private ProxyJDBCExecuteCallback(final SQLType sqlType, final boolean isExceptionThrown, final Map<String, Object> dataMap, final boolean isReturnGeneratedKeys) {
-            super(DatabaseType.MySQL, sqlType, isExceptionThrown, dataMap);
+        private ProxyJDBCExecuteCallback(final SQLType sqlType, final boolean isExceptionThrown, final boolean isReturnGeneratedKeys) {
+            super(DatabaseType.MySQL, sqlType, isExceptionThrown);
             this.isReturnGeneratedKeys = isReturnGeneratedKeys;
         }