Spring项目长事务对并发的影响和事务相关的SQL

何为长事务？

        在项目中某个方法，长时间未提交的事务就可以称为长事务。《=== 操作的数据太多，业务涉及的表比较多，同时存在增删改查，其他操作过多(非DB操作)

        数据操作得很多，比如在一个事务里面插入了很多数据，那么这个事务执行时间自然就会变得很长。
        锁的竞争大，当所有的连接都同时对同一个数据进行操作，那么就会出现排队等待，事务时间自然就会变长。
        事务中有其他非DB操作，比如一些RPC请求，有些人说我的RPC很快的，不会增加事务的运行时间，但是RPC请求本身就是一个不稳定的因素，受很多因素影响，网络波动，下游服务响应缓慢，如果这些因素一旦出现，就会有大量的事务时间很长，有可能导致Mysql挂掉，从而引起雪崩。

缺点        

• 并发情况下，数据库连接池容易被撑爆
• 锁定太多的数据，造成大量的阻塞和锁超时
• 执行时间长，容易造成主从延迟
• 回滚所需要的时间比较长
• undo log膨胀

通常插入少量数据耗时大概5ms，这个可以在业务中打log来评估

SELECT* 
FROMinformation_schema.innodb_trx 
WHERETIME_TO_SEC( timediff( now( ), trx_started ) ) > 10

如何避免大事务

• 在一个事务里面， 避免一次处理太多数据。
• 在一个事务里面，尽量避免不必要的查询，在事务外准备数据。
• 在一个事务里面， 避免耗时太多的操作，造成事务超时。
• 在一个事务里面，一些非DB的操作，比如rpc调用，消息队列的操作尽量放到事务之外操作
• 只读的操作不要加事务控制

 

• 在InnoDB事务中，行锁是在需要的时候才加上的，但并不是不需要了就立刻释放，而是要等到事务结束时才释放。如果你的事务中需要锁多个行，要把最可能造成锁冲突、最可能影响并发度的锁尽量往后放
• 通过SETMAX_EXECUTION_TIME命令， 来控制每个语句查询的最长时间，避免单个语句意外查询太长时间
• 监控 information_schema.Innodb_trx表，设置长事务阈值，超过就报警/或者kill
• 在业务功能测试阶段要求输出所有的general_log，分析日志行为提前发现问题
• 设置innodb_undo_tablespaces值，将undo log分离到独立的表空间。如果真的出现大事务导致回滚段过大，这样设置后清理起来更方便

参考https://blog.csdn.net/li563868273/article/details/103080673?spm=1001.2101.3001.6650.3&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7ECTRLIST%7Edefault-3.no_search_link&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7ECTRLIST%7Edefault-3.no_search_link

基本上写请求的都需要使用事务，而Spring对于事务的使用又特别的简单，只需要一个@Transactional注解即可，如下面的例子：

    @Transactional
    public int createOrder(Order order){
        orderDbStorage.save(order);
        orderItemDbStorage.save(order.getItems());
        return order.getId();
    }

在我们创建订单的时候, 通常需要将订单和订单项放在同一个事务里面保证其满足ACID，这里我们只需要在我们创建订单的方法上面写上事务注解即可。

事务的合理使用
对于上面的创建订单的代码，如果现在需要新增一个需求，在创建订单之后发送一个消息到消息队列或者调用一个RPC，你会怎么做呢？很多同学首先会想到，直接在事务方法里面进行调用：

    @Transactional
    public int createOrder(Order order){
        orderDbStorage.save(order);
        orderItemDbStorage.save(order.getItems());
        sendRpc();
        sendMessage();
        return order.getId();
    }

        事务中嵌套rpc，嵌套一些其他非DB的操作，一般情况下这么写的确也没什么问题，一旦非DB写操作出现比较慢，或者流量比较大，就会出现大事务的问题。由于事务的一直不提交，就会导致数据库连接被占用。这个时候你可能会问，我扩大点数据库连接不就行了吗，100个不行就上1000个，连接池过大会导致线程频繁切换消耗CPU！

那我们应该怎么对其进行优化呢？在这里可以仔细想想，我们的非db操作，其实是不满足我们事务的ACID的，那么干嘛要写在事务里面，所以这里我们可以将其提取出来。

    public int createOrder(Order order){
        createOrderService.createOrder(order);
        sendRpc();
        sendMessage();
    }

        createOrderService，自己注入自己，再调用有事务的方法创建订单，然后在去调用其他非DB操作。如果我们现在想要更复杂一点的逻辑，比如创建订单成功就发送成功的RPC请求，失败就发送失败的RPC请求，由上面的代码我们可以做如下转化：

    public int createOrder(Order order){
        try {
            createOrderService.createOrder(order);
            sendSuccessedRpc();
        }catch (Exception e){
            sendFailedRpc();
            throw e;
        }
    }

通常我们会捕获异常，或者根据返回值来进行一些特殊处理，这里的实现需要显示的捕获异常，并且在次抛出，这种方式不是很优雅，那么怎么才能更好的写这种话逻辑呢？

TransactionSynchronizationManager
在Spring的事务中刚好提供了一些工具方法，来帮助我们完成这种需求。在TransactionSynchronizationManager中提供了让我们对事务注册callBack的方法：

public static void registerSynchronization(TransactionSynchronization synchronization)
            throws IllegalStateException {

        Assert.notNull(synchronization, "TransactionSynchronization must not be null");
        if (!isSynchronizationActive()) {
            throw new IllegalStateException("Transaction synchronization is not active");
        }
        synchronizations.get().add(synchronization);
    }


TransactionSynchronization也就是我们事务的callBack，提供了一些扩展点给我们：

public interface TransactionSynchronization extends Flushable {

    int STATUS_COMMITTED = 0;
    int STATUS_ROLLED_BACK = 1;
    int STATUS_UNKNOWN = 2;
    
    /**
     * 挂起时触发
     */
    void suspend();

    /**
     * 挂起事务抛出异常的时候 会触发
     */
    void resume();

    @Override
    void flush();

    /**
     * 在事务提交之前触发
     */
    void beforeCommit(boolean readOnly);

    /**
     * 在事务完成之前触发
     */
    void beforeCompletion();

    /**
     * 在事务提交之后触发
     */
    void afterCommit();

    /**
     * 在事务完成之后触发
     */
    void afterCompletion(int status);
}

我们可以利用afterComplettion方法实现我们上面的业务逻辑：

    @Transactional
    public int createOrder(Order order){
        orderDbStorage.save(order);
        orderItemDbStorage.save(order.getItems());
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            @Override
            public void afterCompletion(int status) {
                if (status == STATUS_COMMITTED){
                    sendSuccessedRpc();
                }else {
                    sendFailedRpc();
                }
            }
        });
        return order.getId();
    }

这里我们直接实现了afterCompletion，通过事务的status进行判断，我们应该具体发送哪个RPC。当然我们可以进一步封装TransactionSynchronizationManager.registerSynchronization将其封装成一个事务的Util，可以使我们的代码更加简洁。

通过这种方式我们不必把所有非DB操作都写在方法之外，这样代码更具有逻辑连贯性，更加易读，并且优雅。

afterCompletion的坑
这个注册事务的回调代码在我们在我们的业务逻辑中经常会出现，比如某个事务做完之后的刷新缓存，发送消息队列，发送通知消息等等，在日常的使用中，大家用这个基本也没出什么问题，但是在打压的过程中，发现了这一块出现了瓶颈，耗时特别久，通过一系列的监测，发现是从数据库连接池获取连接等待的时间较长，最终我们定位到了afterCompeltion这个动作，居然没有归还数据库连接。

在Spring的AbstractPlatformTransactionManager中，对commit处理的代码如下：

private void processCommit(DefaultTransactionStatus status) throws TransactionException {
        try {
            boolean beforeCompletionInvoked = false;
            try {
                prepareForCommit(status);
                triggerBeforeCommit(status);
                triggerBeforeCompletion(status);
                beforeCompletionInvoked = true;
                boolean globalRollbackOnly = false;
                if (status.isNewTransaction() || isFailEarlyOnGlobalRollbackOnly()) {
                    globalRollbackOnly = status.isGlobalRollbackOnly();
                }
                if (status.hasSavepoint()) {
                    if (status.isDebug()) {
                        logger.debug("Releasing transaction savepoint");
                    }
                    status.releaseHeldSavepoint();
                }
                else if (status.isNewTransaction()) {
                    if (status.isDebug()) {
                        logger.debug("Initiating transaction commit");
                    }
                    doCommit(status);
                }
                // Throw UnexpectedRollbackException if we have a global rollback-only
                // marker but still didn't get a corresponding exception from commit.
                if (globalRollbackOnly) {
                    throw new UnexpectedRollbackException(
                            "Transaction silently rolled back because it has been marked as rollback-only");
                }
            }
    

            // Trigger afterCommit callbacks, with an exception thrown there
            // propagated to callers but the transaction still considered as committed.
            try {
                triggerAfterCommit(status);
            }
            finally {
                triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
            }

        }
        finally {
            cleanupAfterCompletion(status);
        }
    }

这里我们只需要关注 倒数几行代码即可，可以发现我们的triggerAfterCompletion，是倒数第二个执行逻辑，当执行完所有的代码之后就会执行我们的cleanupAfterCompletion，而我们的归还数据库连接也在这段代码之中，这样就导致了我们获取数据库连接变慢。

如何优化
对于上面的问题如何优化呢？这里有三种方案可以进行优化：

将非DB操作提到事务之外，这种方法也就是我们上面最原始的方法，对于一些简单的逻辑可以提取，但是对于一些复杂的逻辑，比如事务的嵌套，嵌套里面调用了afterCompletion，这样做会增大很多工作量，并且很容易出现问题。
通过多线程异步去做，提升数据库连接池归还速度，这种适合于注册afterCompletion时写在事务最后的时候，直接将需要做的放在其它线程去做。但是如果注册afterCompletion的时候出现在我们事务之间，比如嵌套事务，就会导致我们要做的后续业务逻辑和事务并行。
模仿Spring事务回调注册，实现新的注解。上面两种方法都有各自的弊端，所以最后我们采用了这种方法，实现了一个自定义注解@MethodCallBack,在使用事务的上面都打上这个注解，然后通过类似的注册代码进行。
   

@Transactional(rollBackFor = Exception.class)
@MethodCallBack
public int createOrder(Order order){orderDbStorage.save(order);orderItemDbStorage.save(order.getItems());MethodCallbackHelper.registerOnSuccess(() -> sendSuccessedRpc());MethodCallbackHelper.registerOnThrowable(throwable -> sendFailedRpc());return order.getId();
}

 

事务相关的SQL

# 查询所有正在运行的事务及运行时间
SELECTt.*,to_seconds( now( ) ) - to_seconds( t.trx_started ) idle_time 
FROMINFORMATION_SCHEMA.INNODB_TRX t# 查询事务详细信息及执行的SQL
SELECTnow( ),( UNIX_TIMESTAMP( now( ) ) - UNIX_TIMESTAMP( a.trx_started ) ) diff_sec,b.id,b.USER,b.HOST,b.db,d.SQL_TEXT 
FROMinformation_schema.innodb_trx aINNER JOIN information_schema.PROCESSLIST b ON a.TRX_MYSQL_THREAD_ID = b.id AND b.command = 'Sleep'INNER JOIN PERFORMANCE_SCHEMA.threads c ON b.id = c.PROCESSLIST_IDINNER JOIN PERFORMANCE_SCHEMA.events_statements_current d ON d.THREAD_ID = c.THREAD_ID;# 查询事务执行过的所有历史SQL记录
SELECTps.id 'PROCESS ID',ps.USER,ps.HOST,esh.EVENT_ID,trx.trx_started,esh.event_name 'EVENT NAME',esh.sql_text 'SQL',ps.time 
FROMPERFORMANCE_SCHEMA.events_statements_history eshJOIN PERFORMANCE_SCHEMA.threads th ON esh.thread_id = th.thread_idJOIN information_schema.PROCESSLIST ps ON ps.id = th.processlist_idLEFT JOIN information_schema.innodb_trx trx ON trx.trx_mysql_thread_id = ps.id 
WHEREtrx.trx_id IS NOT NULL AND ps.USER != 'SYSTEM_USER' 
ORDER BYesh.EVENT_ID;# 简单查询事务锁select * from sys.innodb_lock_waits# 查询事务锁详细信息SELECTtmp.*,c.SQL_Text blocking_sql_text,p.HOST blocking_host 
FROM(
SELECTr.trx_state wating_trx_state,r.trx_id waiting_trx_id,r.trx_mysql_thread_Id waiting_thread,r.trx_query waiting_query,b.trx_state blocking_trx_state,b.trx_id blocking_trx_id,b.trx_mysql_thread_id blocking_thread,b.trx_query blocking_query 
FROMinformation_schema.innodb_lock_waits wINNER JOIN information_schema.innodb_trx b ON b.trx_id = w.blocking_trx_idINNER JOIN information_schema.innodb_trx r ON r.trx_id = w.requesting_trx_id ) tmp,information_schema.PROCESSLIST p,PERFORMANCE_SCHEMA.events_statements_current c,PERFORMANCE_SCHEMA.threads t 
WHEREtmp.blocking_thread = p.id AND t.thread_id = c.THREAD_ID AND t.PROCESSLIST_ID = p.id