queryTimeout对Cobar不生效的原因

客官°小女子只卖身不卖艺 2022-09-26 03:09 168阅读 0赞

假设我们想对某一条sql做超时限制，我们可能会采用如下的方式：

public static void main(String[] args) {
            ApplicationContext  ctx = SpringApplication.run(SpringBootDemoApplication.class, args);
            JdbcTemplate jdbcTemplate = ctx.getBean(JdbcTemplate.class);
            jdbcTemplate.setQueryTimeout(1);
            jdbcTemplate.execute("select * from `order` where id = 49320135 for update");
        }

这种方式在连接普通`mysql`实例的时候是没有问题的。但是在连接`cobar`时配置`queryTimeout`是不生效的，这是为什么呢？

刚碰到这个问题时我是很迷惑的。因为一般来讲超时本身应该是在客户端做控制的，和服务端(`mysql`实例或者`cobar`实例)是没有关系的。但是现象确实存在，那么就只能调试源码了，也许设计者和我的想法不太一样呢

直接从`JdbcTemplate.execute`出发：

public void execute(final String sql) throws DataAccessException {
            if (logger.isDebugEnabled()) {
                logger.debug("Executing SQL statement [" + sql + "]");
            }
            class ExecuteStatementCallback implements StatementCallback<Object>, SqlProvider { 
                @Override
                public Object doInStatement(Statement stmt) throws SQLException {
                    stmt.execute(sql);
                    return null;
                }
                @Override
                public String getSql() {
                    return sql;
                }
            }
            execute(new ExecuteStatementCallback());
        }

其中`doInStatement(Statement stmt)`方法中的参数`Statement`的实现类是`StatementImpl`，我们重点看下这个类，因为`queryTimeout`实际上最终就是设置到`Statement`层面的，我们看看其中`executeQuery`方法：

public java.sql.ResultSet executeQuery(String sql) throws SQLException {
            synchronized (checkClosed().getConnectionMutex()) {
                // 省略部分代码...
    
                CancelTask timeoutTask = null;
    
                String oldCatalog = null;
    
                try {
                    if (locallyScopedConn.getEnableQueryTimeouts() && this.timeoutInMillis != 0 && locallyScopedConn.versionMeetsMinimum(5, 0, 0)) {
                        // 这里用到了timeout参数，并启动了一个timeoutTask，等下会重点看下
                        timeoutTask = new CancelTask(this);
                        locallyScopedConn.getCancelTimer().schedule(timeoutTask, this.timeoutInMillis);
                    }
    
                    if (!locallyScopedConn.getCatalog().equals(this.currentCatalog)) {
                        oldCatalog = locallyScopedConn.getCatalog();
                        locallyScopedConn.setCatalog(this.currentCatalog);
                    }
    
                    //
                    // Check if we have cached metadata for this query...
                    //
    
                    Field[] cachedFields = null;
    
                    if (locallyScopedConn.getCacheResultSetMetadata()) {
                        cachedMetaData = locallyScopedConn.getCachedMetaData(sql);
    
                        if (cachedMetaData != null) {
                            cachedFields = cachedMetaData.fields;
                        }
                    }
    
                    locallyScopedConn.setSessionMaxRows(this.maxRows);
    
                    statementBegins();
    
                    this.results = locallyScopedConn.execSQL(this, sql, this.maxRows, null, this.resultSetType, this.resultSetConcurrency, doStreaming,
                            this.currentCatalog, cachedFields);
    
                    if (timeoutTask != null) {
                        if (timeoutTask.caughtWhileCancelling != null) {
                            throw timeoutTask.caughtWhileCancelling;
                        }
    
                        timeoutTask.cancel();
    
                        locallyScopedConn.getCancelTimer().purge();
    
                        timeoutTask = null;
                    }
    
                    synchronized (this.cancelTimeoutMutex) {
                        if (this.wasCancelled) {
                            SQLException cause = null;
    
                            if (this.wasCancelledByTimeout) {
                                cause = new MySQLTimeoutException();
                            } else {
                                cause = new MySQLStatementCancelledException();
                            }
    
                            resetCancelledState();
    
                            throw cause;
                        }
                    }
                } finally {
                    this.statementExecuting.set(false);
    
                    if (timeoutTask != null) {
                        timeoutTask.cancel();
    
                        locallyScopedConn.getCancelTimer().purge();
                    }
    
                    if (oldCatalog != null) {
                        locallyScopedConn.setCatalog(oldCatalog);
                    }
                }
    
                this.lastInsertId = this.results.getUpdateID();
    
                if (cachedMetaData != null) {
                    locallyScopedConn.initializeResultsMetadataFromCache(sql, cachedMetaData, this.results);
                } else {
                    if (this.connection.getCacheResultSetMetadata()) {
                        locallyScopedConn.initializeResultsMetadataFromCache(sql, null /* will be created */, this.results);
                    }
                }
    
                return this.results;
            }
        }

可以看到如果`queryTimeout`大于0，那么就会启动一个`CancelTask`：

public void run() {
    
            Thread cancelThread = new Thread() {
    
                @Override
                public void run() {
    
                    Connection cancelConn = null;
                    java.sql.Statement cancelStmt = null;
    
                    try {
                        if (StatementImpl.this.connection.getQueryTimeoutKillsConnection()) {
                            CancelTask.this.toCancel.wasCancelled = true;
                            CancelTask.this.toCancel.wasCancelledByTimeout = true;
                            StatementImpl.this.connection.realClose(false, false, true,
                                    new MySQLStatementCancelledException(Messages.getString("Statement.ConnectionKilledDueToTimeout")));
                        } else {
                            synchronized (StatementImpl.this.cancelTimeoutMutex) {
                                if (CancelTask.this.origConnURL.equals(StatementImpl.this.connection.getURL())) {
                                    //All's fine
                                    cancelConn = StatementImpl.this.connection.duplicate();
                                    cancelStmt = cancelConn.createStatement();
                                    cancelStmt.execute("KILL QUERY " + CancelTask.this.connectionId);
                                } else {
                                    try {
                                        cancelConn = (Connection) DriverManager.getConnection(CancelTask.this.origConnURL, CancelTask.this.origConnProps);
                                        cancelStmt = cancelConn.createStatement();
                                        cancelStmt.execute("KILL QUERY " + CancelTask.this.connectionId);
                                    } catch (NullPointerException npe) {
                                        //Log this? "Failed to connect to " + origConnURL + " and KILL query"
                                    }
                                }
                                CancelTask.this.toCancel.wasCancelled = true;
                                CancelTask.this.toCancel.wasCancelledByTimeout = true;
                            }
                        }
                    } catch (SQLException sqlEx) {
                        CancelTask.this.caughtWhileCancelling = sqlEx;
                    } catch (NullPointerException npe) {
                    } finally {
                        if (cancelStmt != null) {
                            try {
                                cancelStmt.close();
                            } catch (SQLException sqlEx) {
                                throw new RuntimeException(sqlEx.toString());
                            }
                        }
    
                        if (cancelConn != null) {
                            try {
                                cancelConn.close();
                            } catch (SQLException sqlEx) {
                                throw new RuntimeException(sqlEx.toString());
                            }
                        }
    
                        CancelTask.this.toCancel = null;
                        CancelTask.this.origConnProps = null;
                        CancelTask.this.origConnURL = null;
                    }
                }
            };
    
            cancelThread.start();
        }

可以看到`CancelTask`大致分为两步。第一步是通过`Kill Query threadId`命令先将数据库的查询线程杀死。然后第二步再将该查询标记为取消。并且通过第一步将该查询`kill`掉之后，那么本身阻塞的查询会立即返回，并抛出`com.mysql.jdbc.exceptions.jdbc4.MySQLQueryInterruptedException`

再来看看`MysqlIO`中`sqlQueryDirect`方法`catch`里的逻辑处理：

if (this.statementInterceptors != null) {
            invokeStatementInterceptorsPost(query, callingStatement, null, false, sqlEx); 
        }
    
        if (callingStatement != null) {
            synchronized (callingStatement.cancelTimeoutMutex) {
                if (callingStatement.wasCancelled) {
                    SQLException cause = null;
    
                    if (callingStatement.wasCancelledByTimeout) {
                        cause = new MySQLTimeoutException();
                    } else {
                        cause = new MySQLStatementCancelledException();
                    }
    
                    callingStatement.resetCancelledState();
    
                    throw cause;
                }
            }
        }
    
        throw sqlEx;

可以看到对于`canceledByTimeout`这种情况的，`mysql-connector`会重新抛出一个`MySQLTimeoutException`的新异常。那么实际上我们应用只需要根据此异常来判断是否为执行超时就可以了。

总结：sql的`queryTimeout`机制是通过`kill query threadId`命令来实现的，而cobar不支持`kill query`命令，那么自然就不生效了