Spring Boot客户端断开时中止Snowflake查询执行方案

问题根因 #

原有逻辑先全量执行完Snowflake查询再操作响应输出流，查询执行阶段没有任何网络IO交互，无法通过输出流的IOException感知客户端断开事件，必须将断开检测逻辑嵌入查询全流程，同时调整查询执行+响应写入的串行模式，才能实现断开即中止的效果。

核心实现思路 #

• 利用Servlet/Spring Web层提供的请求断开监听能力，在请求开始时就注册断开回调，不依赖输出流写入触发异常判断连接状态
• 保留Snowflake查询的Statement引用，一旦收到断开信号立刻调用Statement.cancel()中断正在执行的查询，释放Snowflake侧计算资源
• 调整结果返回逻辑：开启Snowflake流式结果集，边拉取查询结果边写入响应流，避免全量加载查询结果到内存，同时通过定期刷流兜底感知连接状态
• 关闭接口层面的响应缓存，确保写入操作直接作用于原始网络输出流

代码示例（Spring MVC Servlet栈） #

import jakarta.servlet.AsyncContext;
import jakarta.servlet.AsyncEvent;
import jakarta.servlet.AsyncListener;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.io.OutputStream;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.sql.SQLException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@RestController
public class SnowflakeQueryController {
    // 生产环境替换为自定义配置的业务线程池，禁止使用无界默认线程池
    private final ExecutorService queryExecutor = Executors.newFixedThreadPool(10);

    @GetMapping("/export/snowflake-data")
    public void streamSnowflakeData(HttpServletRequest request, HttpServletResponse response) throws IOException {
        // 开启异步请求，避免阻塞Servlet容器主线程
        AsyncContext asyncContext = request.startAsync();
        // 按业务需求设置异步请求超时时间，0为不启用容器级超时
        asyncContext.setTimeout(0);
        response.setContentType("text/csv;charset=utf-8");
        OutputStream rawOutputStream = response.getOutputStream();

        queryExecutor.submit(() -> {
            try (
                    // 替换为实际业务中从Snowflake数据源获取连接的逻辑
                    Connection snowflakeConn = getSnowflakeConnection();
                    Statement queryStmt = snowflakeConn.createStatement()
            ) {
                // 注册客户端断开、超时、异常事件监听器
                asyncContext.addListener(new AsyncListener() {
                    @Override
                    public void onComplete(AsyncEvent event) {}

                    @Override
                    public void onTimeout(AsyncEvent event) {
                        cancelRunningQuery(queryStmt);
                    }

                    @Override
                    public void onError(AsyncEvent event) {
                        // 客户端关闭浏览器、网络中断都会触发该回调
                        cancelRunningQuery(queryStmt);
                    }

                    @Override
                    public void onStartAsync(AsyncEvent event) {}
                });

                // 配置Snowflake流式拉取，按批次拉取结果避免全量加载到内存
                queryStmt.setFetchSize(1000);
                try (ResultSet resultSet = queryStmt.executeQuery("替换为实际业务的慢查询SQL")) {
                    // 边拉取结果边写入响应流，不要等查询全量返回再写
                    while (resultSet.next()) {
                        // 替换为实际业务的结果序列化逻辑，比如拼接CSV行、JSON片段
                        byte[] rowContent = (resultSet.getString(1) + "\n").getBytes();
                        rawOutputStream.write(rowContent);
                        // 定期刷流，兜底感知网络连接状态
                        rawOutputStream.flush();
                    }
                }
            } catch (Exception e) {
                // 过滤主动取消查询的异常，不需要打印错误日志
                if (!(e instanceof SQLException && "0A000".equals(((SQLException) e).getSQLState()))) {
                    // 其余业务异常按原有逻辑处理，比如返回错误信息
                    e.printStackTrace();
                }
            } finally {
                // 完成异步请求，关闭输出流
                asyncContext.complete();
                try {
                    rawOutputStream.close();
                } catch (IOException ignored) {}
            }
        });
    }

    /**
     * 安全取消正在运行的Snowflake查询
     */
    private void cancelRunningQuery(Statement stmt) {
        try {
            if (!stmt.isClosed()) {
                stmt.cancel();
            }
        } catch (SQLException ignored) {}
    }

    /**
     * 获取Snowflake连接，示例方法，替换为实际业务逻辑
     */
    private Connection getSnowflakeConnection() throws SQLException {
        return null;
    }
}

注意事项 #

• 如果项目中存在全局响应包装、响应缓存逻辑（比如ResponseBodyAdvice统一封装返回值、过滤器缓存响应内容），需要对该接口放行，确保写入的是原始响应输出流，否则写入操作不会触发真实网络IO，无法感知连接状态。
• Snowflake JDBC的Statement.cancel()是线程安全方法，可以直接在监听器回调线程调用，调用后Snowflake服务端会立刻终止对应查询，不会继续占用计算资源。
• 如果使用Spring WebFlux栈，实现逻辑更简化：将Snowflake结果集读取包装为响应式Flux流，在doOnCancel回调中执行查询取消逻辑即可，WebFlux框架会在客户端断开时自动触发取消信号。
• AsyncListener的onError回调作为断开判断的主逻辑，输出流写入时的IOException捕获作为兜底，不要单独依赖IO异常判断断开，避免因容器、代理层的缓存机制导致异常触发延迟。

内容的提问来源于stack exchange，提问作者Sudha Rani S Sudha