TCC-transaction源码(二)：事务恢复

深碍√TFBOYSˉ_ 2023-03-04 13:26 4阅读 0赞

### 一、为什么需要恢复任务 ###

为了处理异常。在TCC事务下，假如A服务调用B服务，B服务超过事务恢复的时间还没有返回，就要取消事务，进行回滚操作，不能让事务一直这么挂着不结束。或者还没等B结果返回A服务就挂了，重启A服务后的遗留事务需要恢复。或者是B返回成功/失败后，A服务执行二阶段的确认提交/回滚事务的方法时候失败了，如果没有异常处理，A的资源便无法释放，事务无法结束。

### 二、恢复任务初始化类 ###

public class RecoverScheduledJob { 
    
        private TransactionRecovery transactionRecovery;
    
        private TransactionConfigurator transactionConfigurator;
    
        private Scheduler scheduler;
    
        public void init() { 
    
            try { 
                MethodInvokingJobDetailFactoryBean jobDetail = new MethodInvokingJobDetailFactoryBean();
                // 任务执行的实体是TransactionRecovery类的startRecover方法
                jobDetail.setTargetObject(transactionRecovery);
                jobDetail.setTargetMethod("startRecover");
                jobDetail.setName("transactionRecoveryJob");
                // 任务不允许并发执行
                jobDetail.setConcurrent(false);
                jobDetail.afterPropertiesSet();
    
                CronTriggerFactoryBean cronTrigger = new CronTriggerFactoryBean();
                cronTrigger.setBeanName("transactionRecoveryCronTrigger");
                // 任务按配置的corn表达式定时执行 
     cronTrigger.setCronExpression(transactionConfigurator.getRecoverConfig().getCronExpression());
                cronTrigger.setJobDetail(jobDetail.getObject());
                cronTrigger.afterPropertiesSet();
    
                scheduler.scheduleJob(jobDetail.getObject(), cronTrigger.getObject());
    
    
                scheduler.start();
    
            } catch (Exception e) { 
                throw new SystemException(e);
            }
        }
    	// ......
    }

SpringTransactionConfigurator是TCC事务的全局配置类，除了提供任务执行的corn表达式，任务其他相关属性外，还有事务异步执行的线程池等。

public class SpringTransactionConfigurator implements TransactionConfigurator { 
    
        private static volatile ExecutorService executorService = null;
    
        @Autowired
        private TransactionRepository transactionRepository;
    	
        /** *DefaultRecoverConfig提供了RecoverConfig接口的默认配置项，包括事务恢复任务执行corn表达式，事务 *恢复间隔时间，恢复最大执行次数； */
        @Autowired(required = false)
        private RecoverConfig recoverConfig = DefaultRecoverConfig.INSTANCE;
    
        private TransactionManager transactionManager;
    	
        /** *配置的线程池是给事务异步提交和异步回滚用的。 *线程工厂创建的线程前缀是 tcc-async-terminate-pool- +线程递增数字 + -thread-，非守护线程，优 *先级5，拒绝策略为当线程池满时通过主线程运行任务 */
        
    public void init() { 
            transactionManager = new TransactionManager();
            transactionManager.setTransactionRepository(transactionRepository);
    
            if (executorService == null) { 
                Executors.defaultThreadFactory();
                synchronized (SpringTransactionConfigurator.class) { 
                    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();
                                        this.group = securityManager != null ? securityManager.getThreadGroup() : Thread.currentThread().getThreadGroup();
                                        this.namePrefix = "tcc-async-terminate-pool-" + poolNumber.getAndIncrement() + "-thread-";
                                    }
    
                                    public Thread newThread(Runnable runnable) { 
                                        Thread thread = new Thread(this.group, runnable, 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);
    
            if (transactionRepository instanceof CachableTransactionRepository) { 
                ((CachableTransactionRepository) transactionRepository).setExpireDuration(recoverConfig.getRecoverDuration());
            }
        }
    	// ...
    }

DefaultRecoverConfig

public class DefaultRecoverConfig implements RecoverConfig { 
    
        public static final RecoverConfig INSTANCE = new DefaultRecoverConfig();
    
        private int maxRetryCount = 30;
    
        private int recoverDuration = 120; //120 seconds
    
        private String cronExpression = "0 */1 * * * ?";
    
        private int asyncTerminateThreadCorePoolSize = 512;
    
        private int asyncTerminateThreadMaxPoolSize = 1024;
    
        private int asyncTerminateThreadWorkQueueSize = 512;
    
        private Set<Class<? extends Exception>> delayCancelExceptions = new HashSet<Class<? extends Exception>>();
    	// 当事务内发生OptimisticLockException和SocketTimeoutException异常的时候，不会立即回滚事务，
        // 预先加载这两个异常
        public DefaultRecoverConfig() { 
            delayCancelExceptions.add(OptimisticLockException.class);
            delayCancelExceptions.add(SocketTimeoutException.class);
        }

### 三、为什么发生SocketTimeoutException和OptimisticLockException异常时不回滚 ###

*  SocketTimeoutException
    
    **作者说：** 不立即回滚，主要考虑是被调用服务方存在一直在正常执行的可能，只是执行的慢，导致了调用方超时，此时如果立即回滚，在被调用方执行cancel操作的同时，被调用方的try方法还在执行，甚至cancel操作执行完了，try方法还没结束，这种情况下业务数据存在不一致的可能。目前解决办法是这类异常不立即回滚，而是由恢复job执行回滚，恢复job会在一段时间后再去调用该被调用方的cancel方法，这个时间可在RecoverConfig中设置，默认120s。
    
    **也就是说try方法还没执行完的时候，就去执行cancel方法，如果try在cancel之后执行完，那就没法恢复了**。
 *  OptimisticLockException
    
    整个TCC框架中，抛出这个异常的地方只有一处，在org.mengyun.tcctransaction.repository.CachableTransactionRepository\#update 方法中，也就是更新事务状态的时候，如果被更新的条数为0，表示这条记录不存在或者记录存在但无法被更新时，抛出这个异常。
    
    出现这个情况的场景有：
    
    1.  A服务调用B服务时创建事务记录，当B很久还没有返回的时候，事务已经到了恢复的时间间隔，A就会取消事务，删除事务记录（B作为事务参与者，取消方法也会被A调用）。之后B再返回的时候，无论返回成功A提交，还是返回失败A回滚，这条事务记录都已经不存在了，就会抛异常。此时已经没有什么好回滚的了，所以忽略这个异常。
    2.  A服务调用B服务时，在真正调用B的方法之前，需要
        
        (1)、创建根事务记录；
        
        (2)、更新根事务的参与者A;
        
        (3)、更新根事务的参与者B;
        
        (4)、。。。。。。如果事务很长，有更多的参与者，则需要诸葛加入事务参与者队列
        
        在这个过程中如果事务就到了恢复间隔，执行回滚删除事务记录，那后面加入事务参与者队列的更新操作，都会抛OptimisticLockException异常。
    
    **场景1**是最常见的抛出OptimisticLockException异常的场景，所以一定要配置事务恢复间隔大于服务调用超时时间。这也符合任务补偿的设计理念，任务是做补偿用的，应该在主干流程执行结束后执行，而不是提前执行。**场景2**出现的概率比较小，毕竟协调事务参与者是用不了多少时间的。

### 四、恢复任务执行体TransactionRecovery ###

public class TransactionRecovery { 
    
        private TransactionConfigurator transactionConfigurator;
    	// 恢复任务执行的方法
        public void startRecover() { 
            List<Transaction> transactions = loadErrorTransactions();
            recoverErrorTransactions(transactions);
        }
    
        private List<Transaction> loadErrorTransactions() { 
            long currentTimeInMillis = Calendar.getInstance().getTimeInMillis();
            TransactionRepository transactionRepository = transactionConfigurator.getTransactionRepository();
            RecoverConfig recoverConfig = transactionConfigurator.getRecoverConfig();
    		/** *获取所有超过了恢复期限的事务，一直跟进方法可以看到是获取事务记录表中 * LAST_UPDATE_TIME < 当前时间 - 恢复间隔的所有记录 */
            return transactionRepository.findAllUnmodifiedSince(new Date(currentTimeInMillis - recoverConfig.getRecoverDuration() * 1000));
        }
    
        private void recoverErrorTransactions(List<Transaction> transactions) { 
    
            for (Transaction transaction : transactions) { 
    			/** * 如果恢复任务执行次数超过了设置的最大恢复次数，打日志跳过这个事务,此时只能人为干预 */
                if (transaction.getRetriedCount() > transactionConfigurator.getRecoverConfig().getMaxRetryCount()) { 
    
                    logger.error(String.format("recover failed with max retry count,will not try again. txid:%s, status:%s,retried count:%d,transaction content:%s", transaction.getXid(), transaction.getStatus().getId(), transaction.getRetriedCount(), JSON.toJSONString(transaction)));
                    continue;
                }
    			// 如果是分支事务，并且当前时间 > 事务发生时间 + 最大恢复次数 * 恢复间隔，也不再执行恢复
                if (transaction.getTransactionType().equals(TransactionType.BRANCH)
                        && (transaction.getCreateTime().getTime() +
                        transactionConfigurator.getRecoverConfig().getMaxRetryCount() *
                                transactionConfigurator.getRecoverConfig().getRecoverDuration() * 1000
                        > System.currentTimeMillis())) { 
                    continue;
                }
                
                try { 
                    // 重试次数+1
                    transaction.addRetriedCount();
    				// 如果事务状态是CONFIRMING，提交事务
                    if (transaction.getStatus().equals(TransactionStatus.CONFIRMING)) { 
    
                        transaction.changeStatus(TransactionStatus.CONFIRMING);
                        transactionConfigurator.getTransactionRepository().update(transaction);
                        transaction.commit();
                       // 提交成功后删除事务记录
                      transactionConfigurator.getTransactionRepository().delete(transaction);
    				// 如果事务状态是CANCELLING，执行回滚
                    } else if (transaction.getStatus().equals(TransactionStatus.CANCELLING)
                            || transaction.getTransactionType().equals(TransactionType.ROOT)) { 
    
                        transaction.changeStatus(TransactionStatus.CANCELLING);
                        transactionConfigurator.getTransactionRepository().update(transaction);
                        transaction.rollback();
                      // 回滚成功后删除事务记录 
                      transactionConfigurator.getTransactionRepository().delete(transaction);
                    }
    
                } catch (Throwable throwable) { 
    
                    if (throwable instanceof OptimisticLockException
                            || ExceptionUtils.getRootCause(throwable) instanceof OptimisticLockException) { 
                        logger.warn(String.format("optimisticLockException happened while recover. txid:%s, status:%s,retried count:%d,transaction content:%s", transaction.getXid(), transaction.getStatus().getId(), transaction.getRetriedCount(), JSON.toJSONString(transaction)), throwable);
                    } else { 
                        logger.error(String.format("recover failed, txid:%s, status:%s,retried count:%d,transaction content:%s", transaction.getXid(), transaction.getStatus().getId(), transaction.getRetriedCount(), JSON.toJSONString(transaction)), throwable);
                    }
                }
            }
        }
    
        public void setTransactionConfigurator(TransactionConfigurator transactionConfigurator) { 
            this.transactionConfigurator = transactionConfigurator;
        }
    }