queryTimeout对Cobar不生效的原因

假设我们想对某一条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命令，那么自然就不生效了

<script type="text/javascript"> $(function () { $('pre.prettyprint code').each(function () { var lines = $(this).text().split('\n').length; var $numbering = $('<ul/>').addClass('pre-numbering').hide(); $(this).addClass('has-numbering').parent().append($numbering); for (i = 1; i <= lines; i++) { $numbering.append($('<li/>').text(i)); }; $numbering.fadeIn(1700); }); }); </script>