添加异步commit，cancel支持

徐烜
1 parent 799dc67f
Showing 14 changed files with 196 additions and 25 deletions
tcc-transaction-core/src/main/java/org/mengyun/tcctransaction/TransactionManager.java
tcc-transaction-core/src/main/java/org/mengyun/tcctransaction/interceptor/CompensableTransactionInterceptor.java
tcc-transaction-core/src/main/java/org/mengyun/tcctransaction/interceptor/ResourceCoordinatorInterceptor.java
tcc-transaction-core/src/main/java/org/mengyun/tcctransaction/interceptor/invoke/MethodContext.java
tcc-transaction-core/src/main/java/org/mengyun/tcctransaction/recover/RecoverConfig.java
tcc-transaction-core/target/classes/org/mengyun/tcctransaction/TransactionManager.class
tcc-transaction-core/target/classes/org/mengyun/tcctransaction/interceptor/CompensableTransactionInterceptor$1.class
tcc-transaction-core/target/classes/org/mengyun/tcctransaction/interceptor/CompensableTransactionInterceptor.class
tcc-transaction-core/target/classes/org/mengyun/tcctransaction/interceptor/ResourceCoordinatorInterceptor.class
tcc-transaction-core/target/classes/org/mengyun/tcctransaction/recover/RecoverConfig.class
tcc-transaction-spring/src/main/java/org/mengyun/tcctransaction/spring/recover/DefaultRecoverConfig.java
tcc-transaction-spring/src/main/java/org/mengyun/tcctransaction/spring/support/TccTransactionConfigurator.java
tcc-transaction-spring/target/classes/org/mengyun/tcctransaction/spring/recover/DefaultRecoverConfig.class
tcc-transaction-spring/target/classes/org/mengyun/tcctransaction/spring/support/TccTransactionConfigurator.class
@@ -10,6 +10,8 @@ import org.mengyun.tcctransaction.repository.TransactionRepository;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import java.util.concurrent.ExecutorService;
+
 /**
  * Created by changmingxie on 10/26/15.
  * 事务管理器：开始事务，提交事务，回滚事务。
@@ -26,6 +28,12 @@ public class TransactionManager {
         this.transactionRepository = transactionRepository;
     }
+    /** 异步执行service */
+    private ExecutorService executorService;
+    public void setExecutorService(ExecutorService executorService) {
+        this.executorService = executorService;
+    }
+
     /** 事务线程局部变量 */
     private ThreadLocal<Transaction> threadLocalTransaction = new ThreadLocal<Transaction>();
     public Transaction getCurrentTransaction() {
@@ -47,14 +55,36 @@ public class TransactionManager {
     /**
      * 提交事务.
      */
-    public void commit() {
+    public void commit(boolean asyncCommit) {
         LOG.debug("==>TransactionManager commit()");
-        Transaction transaction = getCurrentTransaction();
+        final Transaction transaction = getCurrentTransaction();
         transaction.changeStatus(TransactionStatus.CONFIRMING);
         LOG.debug("==>TransactionManager update transaction status to CONFIRMING");
         transactionRepository.update(transaction);
+        if (asyncCommit) {
+            try {
+                Long statTime = System.currentTimeMillis();
+                // TODO：后面会对submit返回的Future做处理
+                executorService.submit(new Runnable() {
+                    @Override
+                    public void run() {
+                        commitTransaction(transaction);
+                    }
+                });
+                LOG.debug("async submit cost time:" + (System.currentTimeMillis() - statTime));
+            } catch (Throwable exp) {
+                LOG.warn("compensable transaction async submit confirm failed, recovery job will try to confirm later.", exp);
+                throw new ConfirmingException(exp);
+            }
+        } else {
+            commitTransaction(transaction);
+        }
+
+
+    }
+    private void commitTransaction(Transaction transaction) {
         try {
             LOG.info("==>TransactionManager transaction begin commit()");
             transaction.commit();
@@ -68,13 +98,31 @@ public class TransactionManager {
     /**
      * 回滚事务.
      */
-    public void rollback() {
+    public void rollback(boolean asyncRollback) {
         LOG.debug("==>TransactionManager rollback()");
-        Transaction transaction = getCurrentTransaction();
+        final Transaction transaction = getCurrentTransaction();
         transaction.changeStatus(TransactionStatus.CANCELLING);
         transactionRepository.update(transaction);
+        if (asyncRollback) {
+            try {
+                executorService.submit(new Runnable() {
+                    @Override
+                    public void run() {
+                        rollbackTransaction(transaction);
+                    }
+                });
+            } catch (Throwable exp) {
+                LOG.warn("compensable transaction async rollback failed, recovery job will try to rollback later.", exp);
+                throw new CancellingException(exp);
+            }
+        } else {
+            rollbackTransaction(transaction);
+        }
+    }
+
+    private void rollbackTransaction(Transaction transaction) {
         try {
             LOG.info("==>TransactionManager transaction begin rollback()");
             transaction.rollback();
@@ -54,20 +54,22 @@ public class CompensableTransactionInterceptor {
     /**
      * 主事务方法的处理.
-     * @param compensableMethodContext
+     * @param methodContext
      * @throws Throwable
      */
-    private Object rootMethodProceed(MethodContext compensableMethodContext) throws Throwable {
+    private Object rootMethodProceed(MethodContext methodContext) throws Throwable {
         LOG.debug("==>rootMethodProceed");
+        boolean asyncConfirm = methodContext.isAsyncConfirm();
+        boolean asyncCancel = methodContext.isAsyncCancel();
         TransactionManager transactionManager = transactionConfigurator.getTransactionManager();
         transactionManager.begin(); // 事务开始（创建事务日志记录，并在当前线程缓存该事务日志记录）
-        Object returnValue = null; // 返回值
+        Object returnValue; // 返回值
         try {
             LOG.debug("==>rootMethodProceed try begin");
-            returnValue = compensableMethodContext.proceed();  // Try (开始执行被拦截的方法)
+            returnValue = methodContext.proceed();  // Try (开始执行被拦截的方法)
             LOG.debug("==>rootMethodProceed try end");
         } catch (OptimisticLockException e) {
@@ -75,25 +77,28 @@ public class CompensableTransactionInterceptor {
             throw e; //do not rollback, waiting for recovery job
         } catch (Throwable tryingException) {
             LOG.warn("compensable transaction trying failed.", tryingException);
-            transactionManager.rollback();
+            transactionManager.rollback(asyncCancel);
             throw tryingException;
         }
         LOG.info("===>rootMethodProceed begin commit()");
-        transactionManager.commit(); // Try检验正常后提交(事务管理器在控制提交)
+        transactionManager.commit(asyncConfirm); // Try检验正常后提交(事务管理器在控制提交)
         return returnValue;
     }
     /**
      * 服务提供者事务方法处理.
-     * @param compensableMethodContext
+     * @param methodContext
      * @throws Throwable
      */
-    private Object providerMethodProceed(MethodContext compensableMethodContext) throws Throwable {
-        TransactionContext transactionContext = compensableMethodContext.getTransactionContext();
+    private Object providerMethodProceed(MethodContext methodContext) throws Throwable {
+        TransactionContext transactionContext = methodContext.getTransactionContext();
+        boolean asyncConfirm = methodContext.isAsyncConfirm();
+        boolean asyncCancel = methodContext.isAsyncCancel();
-        LOG.debug("==>providerMethodProceed transactionStatus:{}", TransactionStatus.valueOf(transactionContext.getStatus()).toString());
+        LOG.debug("==>providerMethodProceed transactionStatus:{}",
+                TransactionStatus.valueOf(transactionContext.getStatus()).toString());
         switch (TransactionStatus.valueOf(transactionContext.getStatus())) {
             case TRYING:
@@ -101,15 +106,15 @@ public class CompensableTransactionInterceptor {
                 // 基于全局事务ID扩展创建新的分支事务，并存于当前线程的事务局部变量中.
                 transactionConfigurator.getTransactionManager().propagationNewBegin(transactionContext);
                 LOG.debug("==>providerMethodProceed try end");
-                return compensableMethodContext.proceed();
+                return methodContext.proceed();
             case CONFIRMING:
                 try {
                     LOG.debug("==>providerMethodProceed confirm begin");
                     // 找出存在的事务并处理.
                     transactionConfigurator.getTransactionManager().propagationExistBegin(transactionContext);
-                    transactionConfigurator.getTransactionManager().commit(); // 提交
+                    transactionConfigurator.getTransactionManager().commit(asyncConfirm); // 提交
                     LOG.debug("==>providerMethodProceed confirm end");
-                } catch (NoExistedTransactionException excepton) {
+                } catch (NoExistedTransactionException exception) {
                     //the transaction has been commit,ignore it.
                 }
                 break;
@@ -117,7 +122,7 @@ public class CompensableTransactionInterceptor {
                 try {
                     LOG.debug("==>providerMethodProceed cancel begin");
                     transactionConfigurator.getTransactionManager().propagationExistBegin(transactionContext);
-                    transactionConfigurator.getTransactionManager().rollback(); // 回滚
+                    transactionConfigurator.getTransactionManager().rollback(asyncCancel); // 回滚
                     LOG.debug("==>providerMethodProceed cancel end");
                 } catch (NoExistedTransactionException exception) {
                     //the transaction has been rollback,ignore it.
@@ -125,7 +130,7 @@ public class CompensableTransactionInterceptor {
                 break;
         }
-        Method method = compensableMethodContext.getMethod();
+        Method method = methodContext.getMethod();
         return ReflectionUtils.getNullValue(method.getReturnType());
     }
@@ -67,7 +67,7 @@ public class ResourceCoordinatorInterceptor {
         }
         LOG.debug("==>pjp.proceed(pjp.getArgs())");
-        return pjp.proceed(pjp.getArgs());
+        return methodContext.proceedWithInArgs();
     }
     /**
@@ -118,6 +118,14 @@ public class MethodContext {
     }
     /**
+     * 是否异步confirm。
+     * @return
+     */
+    public boolean isAsyncConfirm() {
+        return this.compensable.asyncConfirm();
+    }
+
+    /**
      * 获取cancel方法名。
      * @return
      */
@@ -126,7 +134,15 @@ public class MethodContext {
     }
     /**
-     * 执行方法。
+     * 是否异步cancel。
+     * @return
+     */
+    public boolean isAsyncCancel() {
+        return this.compensable.asyncCancel();
+    }
+
+    /**
+     * 执行方法（切入点定义的是标注）。
      * @return
      * @throws Throwable
      */
@@ -134,5 +150,14 @@ public class MethodContext {
         return this.proceedingJoinPoint.proceed();
     }
+    /**
+     * 执行方法（切入点定义的是方法签名）
+     * @return
+     * @throws Throwable
+     */
+    public Object proceedWithInArgs() throws Throwable {
+        return this.proceedingJoinPoint.proceed(this.getProceedingJoinPoint().getArgs());
+    }
+
 }
@@ -22,4 +22,23 @@ public interface RecoverConfig {
      * @return
      */
     String getCronExpression();
+
+    /**
+     * 获取线程池中核心线程数的最大值.
+     * @return
+     */
+    int getAsyncTerminateThreadCorePoolSize();
+
+    /**
+     * 获取线程池中能拥有最多线程数。
+     * @return
+     */
+    int getAsyncTerminateThreadMaxPoolSize();
+
+    /**
+     * 获取用于缓存任务的阻塞队列。
+     * @return
+     */
+    int getAsyncTerminateThreadWorkQueueSize();
+
 }
@@ -21,6 +21,28 @@ public class DefaultRecoverConfig implements RecoverConfig {
     /** 恢复Job触发间隔配置，默认是(每分钟) */
     private String cronExpression = "0 */1 * * * ?";
+    /** 线程池中核心线程数的最大值 */
+    private int asyncTerminateThreadCorePoolSize = 512;
+    /** 线程池中能拥有最多线程数 */
+    private int asyncTerminateThreadMaxPoolSize = 1024;
+    /** 用于缓存任务的阻塞队列 */
+    private int asyncTerminateThreadWorkQueueSize = 512;
+
+    @Override
+    public int getAsyncTerminateThreadCorePoolSize() {
+        return asyncTerminateThreadCorePoolSize;
+    }
+
+    @Override
+    public int getAsyncTerminateThreadMaxPoolSize() {
+        return asyncTerminateThreadMaxPoolSize;
+    }
+
+    @Override
+    public int getAsyncTerminateThreadWorkQueueSize() {
+        return asyncTerminateThreadWorkQueueSize;
+    }
+
     @Override
     public int getMaxRetryCount() {
         return maxRetryCount;
@@ -9,6 +9,8 @@ import org.mengyun.tcctransaction.support.TransactionConfigurator;
 import org.springframework.beans.factory.annotation.Autowired;
 import javax.annotation.PostConstruct;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
 /**
  * Created by changmingxie on 11/11/15.
@@ -32,15 +34,16 @@ public class TccTransactionConfigurator implements TransactionConfigurator {
         return recoverConfig;
     }
-    /**
-     * 根据事务配置器创建事务管理器.
-     */
+    /** 事务管理器 */
     private TransactionManager transactionManager = new TransactionManager();
     @Override
     public TransactionManager getTransactionManager() {
         return transactionManager;
     }
+    /** 线程池配置 */
+    private static volatile ExecutorService executorService;
+
     @PostConstruct
     public void init() {
         // 属性注入后调用
@@ -52,10 +55,59 @@ public class TccTransactionConfigurator implements TransactionConfigurator {
                     recoverConfig.getRecoverDuration()
             );
         }
+
         // 2、事务配置器配置事务repo
         transactionManager.setTransactionRepository(transactionRepository);
-        // TODO：3、executeService配置
+        // 3、线程池配置
+        if (executorService == null) {
+            executorService = new ThreadPoolExecutor(
+                    recoverConfig.getAsyncTerminateThreadCorePoolSize(), // 核心线程数
+                    recoverConfig.getAsyncTerminateThreadMaxPoolSize(), // 最大线程数
+                    5L, // 最大空闲时间
+                    TimeUnit.SECONDS, // 时间单位（秒）
+                    new ArrayBlockingQueue<Runnable>( // 缓存任务的阻塞队列
+                            recoverConfig.getAsyncTerminateThreadWorkQueueSize()),
+                    new ThreadFactory() { // 创建线程的工厂
+                        final AtomicInteger poolNumber = new AtomicInteger(1);
+                        final ThreadGroup group;
+                        final AtomicInteger threadNumber = new AtomicInteger(1);
+                        final String namePrefix;
+
+                        {
+                            SecurityManager securityManager = System.getSecurityManager();
+                            if (securityManager == null) {
+                                this.group = Thread.currentThread().getThreadGroup();
+                            } else {
+                                this.group = securityManager.getThreadGroup();
+                            }
+                            this.namePrefix = "tcc-async-terminate-pool-" + poolNumber.getAndIncrement() + "-thread-";
+                        }
+
+                        @Override
+                        public Thread newThread(Runnable r) {
+                            Thread thread = new Thread(
+                                    this.group,
+                                    r,
+                                    this.namePrefix + this.threadNumber.getAndIncrement(),
+                                    0L
+                            );
+                            if (thread.isDaemon()) {
+                                thread.setDaemon(false);
+                            }
+
+                            if (thread.getPriority() != 5) {
+                                thread.setPriority(5);
+                            }
+
+                            return thread;
+                        }
+                    },
+                    new ThreadPoolExecutor.CallerRunsPolicy() // 线程池满了且任务队列也满了，由向线程池提交任务的线程来执行该任务
+            );
+        }
+        transactionManager.setExecutorService(executorService);
+
     }