如何通过JDBC取消MySQL 8中长时间运行的查询？

问题背景 #

我通过JDBC操作MySQL 8数据库，执行一个可能运行极长时间的查询，需要在触发特定条件（如外部信号、数据处理错误等）时终止该查询。但测试后发现statement.cancel()完全不生效，想解决以下问题：

• 如何实现查询的可靠取消？是否只能通过另一个连接执行KILL <pid>？
• 如果采用KILL方式，怎么可靠识别目标长查询的PID？
• 有没有更稳健的实现方案？

原无效尝试代码 #

最初尝试用stmt.cancel()取消查询，但执行后程序卡住，控制台仅输出到"Statement is not yet closed"，后续代码无输出：

val data = produce<List<Data>> {
    db.connection.use { conn ->
        conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY).use { stmt ->
            stmt.fetchSize = Int.MIN_VALUE
            stmt.executeQuery("...").use { rs ->
                var batch = mutableListOf<Data>()
                while (rs.next()) {
                    // 处理结果集
                    val data = Data(
                        id = rs.getLong("id"),
                        subAccountId = rs.getInt("sub_account_id")
                    )
                    batch.add(data)
                    if (batch.size >= 100) {
                        println("Sending batch...")
                        send(batch)
                        if (done) {
                            println("Done, closing...")
                            close()
                            break
                        }
                        batch = mutableListOf()
                    }
                }
                println("Out of loop, cancelling statement")
                synchronized(this) {
                    println("In synchronized")
                    if (!stmt.isClosed) {
                        println("Statement is not yet closed") // 会打印
                        stmt.cancel()
                        stmt.close()
                        conn.close()
                    }
                }
            }
            println("Out of statement") // 不打印
        }
        println("Out of connection") // 不打印
    }
    println("End of production...") // 不打印
}

可行实现方案 #

参考相关论坛讨论后，我实现了基于KILL QUERY的取消方案，以下代码测试后三种取消方式均生效：

@OptIn(ExperimentalCoroutinesApi::class, DelicateCoroutinesApi::class)
fun dbSqlKillTestWithCoroutines() {
    runBlocking {
        db.connection.use { conn ->
            conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY).use { stmt ->
                stmt.fetchSize = Int.MIN_VALUE

                // 获取当前连接ID
                val connectionId = stmt.executeQuery("SELECT CONNECTION_ID()").use { rs ->
                    rs.next()
                    rs.getString(1)
                }

                // 生产数据：分批读取查询结果
                val data = produce<List<Data>>(Dispatchers.IO) {
                    stmt.executeQuery("SELECT ...").use { rs ->
                        var batch = mutableListOf<Data>()
                        while (rs.next()) {
                            val data = Data(...) // 结果集映射逻辑
                            batch.add(data)
                            if (batch.size >= 100) {
                                println("Sending batch")
                                send(batch)
                                batch = mutableListOf()
                            }
                        }
                    }
                }

                // 消费数据，触发取消条件
                var count = 0
                for (batch in data) {
                    count++
                    println("Received batch $count")
                    if (count >= 3) {
                        println("Attempting to cancel query...")
                        println("Breaking...")
                        break
                    }
                }

                // 方式3：同一上下文执行KILL（测试生效）
                // println("Breaking in same context...")
                // db.connection.use { conn2 ->
                //     conn2.createStatement().use { stmt2 ->
                //         stmt2.execute("KILL QUERY $connectionId")
                //     }
                // }

                // 方式2：用协程创建单独线程执行KILL
                // newSingleThreadContext("ctx").use { ctx ->
                //     launch(ctx) {
                //         println("Breaking in single thread context...")
                //         db.connection.use { conn2 ->
                //             conn2.createStatement().use { stmt2 ->
                //                 stmt2.execute("KILL QUERY $connectionId")
                //             }
                //         }
                //     }
                // }

                // 方式1：手动创建线程执行KILL
                thread(true) {
                    println("Breaking in thread...")
                    db.connection.use { conn2 ->
                        conn2.createStatement().use { stmt2 ->
                            stmt2.execute("KILL QUERY $connectionId")
                        }
                    }
                }
            }
        }
    }
}

注：之前看到资料说KILL QUERY需要在单独线程执行，但测试发现同一上下文执行也能生效，推测可能是特定条件下的例外。

目前该方案能满足需求：支持超长时间SELECT查询、分批处理避免内存溢出、可触发终止并后续断点恢复，但希望得到更稳健的优化建议。

内容的提问来源于stack exchange，提问作者nathlrowe