MySQL JDBC流式查询Socket关闭问题排查与解决方案咨询

通过com.mysql.cj.jdbc.MysqlDataSource连接MySQL只读副本库，并行运行约800个返回1万-20万行的长时中大型查询，协程上限设为64个。为避免OOM，每个查询在独立线程使用新连接，且未设置queryTimeout。结果处理逻辑可能通过Hikari连接池执行主库查询，已确保嵌套数据库访问在独立线程中，无事务共享。

但每执行25-30次调用时，会在快速处理行数据阶段抛出java.io.IOException: Socket is closed异常，线程日志无明显时间间隔。目前无法在支持幂等重启或游标恢复的栈帧中捕获该异常，现咨询以下问题：

1. 异常可能的原因
2. 重试或校验的可行方案
3. 如何确保socketTimeout设置为0

流式查询代码 #

database.transactionManager.newConnection().use { connection ->
    connection.isReadOnly = true
    database.dirtyExecuteExpression(expression) { statement ->
        logger.debug { "Running query: $statement" }
        if (database.dialect is MySqlDialect) statement.fetchSize = Integer.MIN_VALUE
        statement.queryTimeout = 0 // no timeout
        try {
            statement.executeQuery().use { rs ->
                logger.debug { "Query: $rs" }
                while (rs.next()) {
                    action(rs)
                }
            }
        } catch (e: SQLException) {
            logger.error(e) { "Error during suspendable query $statement on ${statement.connection}" }
            throw e
        }
    }
}

/**
 * Sets `transactionIsolation` to [Connection.TRANSACTION_READ_UNCOMMITTED] which means no locking, but is
 * otherwise a copy of [Database.executeExpression]
 */
inline fun <T> Database.dirtyExecuteExpression(expression: SqlExpression, func: (PreparedStatement) -> T): T {
    val (sql, args) = formatExpression(expression)

    // the pattern is to use a non-pooled DB if the body is likely to do db queries.
    useConnection { conn ->
        conn.transactionIsolation = Connection.TRANSACTION_READ_UNCOMMITTED
        // see https://dev.mysql.com/doc/connector-j/en/connector-j-reference-implementation-notes.html
        conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY).use { statement ->
            statement.setArguments(args)
            return func(statement)
        }
    }
}

1. 异常可能的原因 #

• MySQL端闲置超时断开：只读副本的wait_timeout或interactive_timeout设置过短，若结果处理过程中存在短暂停顿（比如主库查询耗时），MySQL会判定连接闲置并主动关闭。
• 连接复用冲突：dirtyExecuteExpression中的useConnection若实际使用了连接池（而非每次创建新连接），会导致多个查询复用同一连接，当某一操作关闭连接时，正在处理结果集的操作就会触发Socket关闭异常。
• 网络设备主动断连：防火墙、负载均衡等中间设备的闲置超时策略，会断开长时间未传输数据的TCP连接，尤其在结果集处理出现间隙时容易触发。
• Connector/J流式查询管理异常：开启fetchSize = Integer.MIN_VALUE流式查询后，若连接被提前回收（比如外部代码意外关闭），会在rs.next()时抛出Socket关闭错误。

2. 重试或校验的可行方案 #

• 幂等化重试：先确认查询和结果处理逻辑是幂等的（重复执行无副作用），在最外层捕获Socket is closed异常，重新创建连接执行查询。
• 连接预校验：执行查询前调用connection.isValid(0)校验连接有效性，无效则重新创建连接（注意该方法会发送轻量查询，需权衡性能）。
• 拆分大查询：将大查询拆分为分页查询（按主键分段），每次处理小批量数据，缩短单连接占用时间，降低被断开的概率。
• 结果处理异步化：把结果集处理逻辑放入独立线程池，避免主查询线程被阻塞，确保MySQL连接持续活跃，减少被判定为闲置的可能。
• 行级异常捕获：在rs.next()循环内捕获异常，记录当前处理行号，后续重试时可通过查询条件过滤已处理行，实现断点续处理。

3. 如何确保socketTimeout设置为0 #

• 直接配置MysqlDataSource：

  val dataSource = MysqlDataSource().apply {
      // 方式1：通过URL参数设置
      setURL("jdbc:mysql://your-replica-host:3306/db_name?socketTimeout=0")
      // 方式2：通过API设置
      socketTimeout = 0
  }
  

• 验证生效参数：创建连接后，通过connection.metaData.connectionURL查看实际URL参数，确认socketTimeout=0存在；或开启com.mysql.cj的DEBUG日志，查看连接建立时的参数配置。
• 覆盖框架默认值：若使用Hikari等连接池，需在池配置中明确设置dataSource.socketTimeout=0，避免被池的默认配置覆盖。

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