绝密笔记 | 基于Druid的长事务监听实现

/**
 * @Project
 * @Description 大事务监控
 * @Date 2022/10/1 下午3:52
 */
public class MyDruidTxMonitorFilter extends FilterEventAdapter {
    //记录事务状态与事务开启时间
    private static final ThreadLocal<Long> TX_BEGIN_TIME = new ThreadLocal<>();
    //记录事务过程中执行的sql（保留执行顺序）
    private static final ThreadLocal<List<String>> TX_SQL_LIST = new ThreadLocal<List<String>>();
 
 
    @Override
    public void connection_setAutoCommit(FilterChain chain, ConnectionProxy connection, boolean autoCommit)
            throws SQLException {
        super.connection_setAutoCommit(chain, connection, autoCommit);
        if (!autoCommit) {
            monitorReady();
        }
    }
 
 
 
 
 
 
    @Override
    protected void statementExecuteBefore(StatementProxy statement, String sql) {
        monitorSql(sql);
    }
    @Override
    protected void statementExecuteUpdateBefore(StatementProxy statement, String sql) {
        monitorSql(sql);
    }
    @Override
    protected void statementExecuteQueryBefore(StatementProxy statement, String sql) {
        monitorSql(sql);
    }
 
 
    @Override
    public void connection_commit(FilterChain chain, ConnectionProxy connection) throws SQLException {
        try {
            super.connection_commit(chain, connection);
            //提交完成后清理本次事务的开始时间、执行的sql等线程绑定的内容
            monitorTransactionTime();
        } finally {
            monitorRemove();
        }
    }
 
 
    @Override
    public void connection_rollback(FilterChain chain, ConnectionProxy connection) throws SQLException {
        try {
            super.connection_rollback(chain, connection);
            //回滚完成后清理本次事务的开始时间、执行的sql等线程绑定的内容
            monitorTransactionTime();
        } finally {
            monitorRemove();
        }
    }
 
 
    @Override
    public void connection_rollback(FilterChain chain, ConnectionProxy connection, Savepoint savepoint)
            throws SQLException {
        try {
            super.connection_rollback(chain, connection, savepoint);
            //回滚完成后清理本次事务的开始时间、执行的sql等线程绑定的内容
            monitorTransactionTime();
        } finally {
            monitorRemove();
        }
    }
 
 
    /**
     * 监控准备
     */
    private void monitorReady() {
        try {
            if (monitorUnable()) return;
            long beginTime = System.currentTimeMillis();
            TX_BEGIN_TIME.set(beginTime);
            LinkedList<String> list = new LinkedList<>();
            list.add("transaction begin time:"+beginTime);
            TX_SQL_LIST.set(list);
        }catch (Throwable e){
            e.printStackTrace();
        }
 
 
    }
 
 
    /**
     * 是否已禁用
     * @return
     */
    private boolean monitorUnable() {
        LongTxMonitor config = SpringUtil.getBean(LongTxMonitor.class);
        if (!config.getHealthCheckLongTxEnable()) {
            return true;
        }
        return false;
    }
 
 
    /**
     * sql记录
     * @param sql
     */
    private void monitorSql(String sql) {
        try {
            if (monitorUnable()) return;
            if (TX_BEGIN_TIME.get() != null) {
                //只记录开启了事务的sql
                TX_SQL_LIST.get().add("current time:"+System.currentTimeMillis()+",");
                TX_SQL_LIST.get().add(sql+";");
            }
        }catch (Throwable e){
            e.printStackTrace();
        }
 
 
    }
    /**
     * 清理
     */
    private void monitorRemove() {
        try {
            if (monitorUnable()) return;
            TX_BEGIN_TIME.remove();
            TX_SQL_LIST.remove();
        }catch (Throwable e){
            e.printStackTrace();
        }
 
 
    }
 
 
    /**
     * 监控事务持续时间
     */
    private void monitorTransactionTime() {
        try {
            if (monitorUnable()) return;
            //1、计算本次事务持续时间  2、长事务告警（记录本次事务涉及到的sql+记录本次事务的应用层调用栈）
            if (TX_BEGIN_TIME.get()!=null) {
                long currentTime = System.currentTimeMillis();
                long timeCost = currentTime - TX_BEGIN_TIME.get().longValue();
 
 
                LongTxMonitor config = SpringUtil.getBean(LongTxMonitor.class);
                if (config.getHealthCheckLongTxEnable()) {
                    if (timeCost>=config.getHealthCheckLongTx()) {
                        //长事务告警
                        CommonEvent.CommonDto commonDto = new CommonEvent.CommonDto();
                        commonDto.setEventTime(LocalDateTime.now().toString());
                        commonDto.setEventType("长事务");
 
 
                        StringBuilder sqls = new StringBuilder("SQL详情:");
                        TX_SQL_LIST.get().forEach((sql)->{sqls.append(sql);});
 
 
                        commonDto.setArgs(sqls.toString());
                        commonDto.setTimeCost(timeCost);
                        commonDto.setLogId(ThreadContext.get(CommonConst.logId));
                        //获取调用方接口名称和方法
                        commonDto.setStackTraceElements(HealthCheckUtil.getCurrentTTKThreadTrace());
                        SpringUtil.getApplicationEventPublisher().publishEvent(new CommonEvent(commonDto));
                    }
 
 
                }
            }
        }catch (Throwable e){
            e.printStackTrace();
        }
 
 
    }
 
 
 
 
}

复制

<bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource" destroy-method="close">
   <property name="url" value="${jdbc.url}" />
   <property name="username" value="${jdbc.username}" />
   <property name="password" value="${jdbc.password}" />
   <property name="proxyFilters">
      <list>
         <bean id="txMonitor" class="com.xxx.MyDruidTxMonitorFilter" />
      </list>
   </property>
</bean>

复制