使用Mono.take设置请求超时引发Stream.closed错误，影响其他请求

问题根源 #

你遇到的Stream.closed错误，本质是第三方阻塞SDK在执行线程被中断时，会关闭共享的底层IO资源（比如连接池中的HTTP连接）。而Reactor的take/timeout默认会触发订阅取消，进而中断执行阻塞调用的线程，导致共享资源被关闭，影响其他复用该资源的请求。

解决方案 #

方案1：阻止超时取消时中断阻塞线程

通过CompletableFuture包装阻塞调用，超时触发时仅标记任务取消但不中断线程，避免SDK因线程中断关闭共享资源。修改工具类代码如下：

public Mono<Response> callSdk(Request request) {
    return Mono.create(sink -> {
        // 使用指定的弹性线程池执行阻塞SDK调用
        CompletableFuture<Response> future = CompletableFuture.supplyAsync(
                () -> blockingSdk(request),
                Schedulers.newBoundedElastic(8, 50000, "call-scheduler")
        );

        // 调度超时任务
        ScheduledFuture<?> timeoutTask = sink.scheduler().schedule(
                () -> {
                    if (!future.isDone()) {
                        // 取消任务但不中断线程，防止SDK关闭共享资源
                        future.cancel(false);
                        sink.error(new TimeoutException("Request not completed in time"));
                    }
                },
                requestTimeoutSeconds,
                TimeUnit.SECONDS
        );

        // 处理任务完成或异常
        future.whenComplete((response, throwable) -> {
            timeoutTask.cancel(false); // 取消未触发的超时任务
            if (throwable != null) {
                // 忽略取消异常（已通过超时逻辑处理）
                if (!(throwable instanceof CancellationException)) {
                    sink.error(throwable);
                }
            } else {
                sink.success(response);
            }
        });

        // 外部取消Mono时，同样不中断线程
        sink.onCancel(() -> future.cancel(false));
    })
    .onErrorResume(Exception.class, this::handleExceptionMono);
}

方案2：隔离SDK资源（若可行）

如果SDK支持实例化而非依赖静态单例，为每个请求创建独立的SDK实例，从根源避免资源共享冲突。此方案可能增加资源开销，需结合SDK特性评估可行性。

补充优化建议 #

• 将原控制器层的subscribeOn移到工具类的阻塞调用执行处，确保阻塞操作始终在指定线程池运行，避免线程泄漏。
• 方案1中未中断的阻塞线程可能会继续执行一段时间，需确保线程池的核心/最大线程数配置合理，避免资源耗尽。

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