如何跟踪异步线程并处理SSE Emitter的管道破裂错误？

针对你遇到的客户端断开连接触发Broken Pipe错误、最终被Tomcat记录为500错误的问题，以下是几个直接可行的解决思路：

1. 给SseEmitter绑定自定义错误回调 #

这是最直接的方案，Spring的SseEmitter提供了onError回调，能捕获异步处理过程中（包括Tomcat写响应时）出现的IO异常，避免错误冒泡到容器层：

修改你的Controller代码：

@GetMapping("/mySSEStream")
public SseEmitter sseEmitter() {
    SseEmitter emitter = new SseEmitter(-1L);
    
    // 绑定错误回调，捕获Broken Pipe等IO异常
    emitter.onError(throwable -> {
        if (throwable instanceof IOException && throwable.getMessage().contains("Broken pipe")) {
            // 客户端断开属于正常场景，无需标记为错误
            log.debug("Client disconnected, broken pipe ignored");
            try {
                emitter.complete();
            } catch (IllegalStateException e) {
                log.debug("Emitter already completed");
            }
        } else {
            // 其他异常按原有逻辑处理
            emitter.completeWithError(throwable);
        }
    });

    MyRunner streamingRunner = new MyRunner(emitter);
    cachedThreadPool.execute(streamingRunner);
    return emitter;
}

这个回调会在WebAsyncManager捕获到异步错误时触发，你可以在这里直接拦截Broken Pipe异常，调用complete()而不是completeWithError()，这样就不会生成500错误日志。

2. 在sendData方法中细粒度捕获异常 #

确保每次调用SseEmitter.send()时都捕获IO异常，并且判断是否为Broken Pipe：

修改MyRunner中的sendData逻辑（假设sendData是循环发送数据）：

private void sendData() throws IOException {
    while (/* 数据发送条件 */) {
        try {
            sseEmitter.send(SseEmitter.event().data("your data"));
            // 按需添加延迟
            Thread.sleep(1000);
        } catch (IOException e) {
            // 判断是否为Broken Pipe异常
            if (isBrokenPipe(e)) {
                log.debug("Client disconnected during send");
                // 直接返回，不抛出异常
                return;
            }
            // 其他IO异常抛出
            throw e;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return;
        }
    }
}

// 辅助方法判断是否为Broken Pipe
private boolean isBrokenPipe(IOException e) {
    // 不同操作系统/容器的错误消息可能不同，可根据实际日志调整
    return e.getMessage() != null && (e.getMessage().contains("Broken pipe") || e.getMessage().contains("EPIPE"));
}

同时调整MyRunner的run方法，避免在Broken Pipe场景下调用completeWithError()：

@Override
public void run() {
    try {
        sendData();
    } catch (IOException ioException) {
        // 只有非Broken Pipe的IO异常才标记错误
        if (!isBrokenPipe(ioException)) {
            sseEmitter.completeWithError(ioException);
        }
    } finally {
        try {
            sseEmitter.complete();
        } catch (IllegalStateException illegalStateException) {
            log.debug("SSE Emitter already closed...");
        }
    }
}

3. 配置Spring WebAsyncManager的异常处理 #

如果上述方案仍不生效，可以通过WebAsyncManager注册全局的异步异常处理器：

创建一个AsyncHandlerInterceptor的实现，在preHandle中注册异常处理器：

public class AsyncExceptionInterceptor implements AsyncHandlerInterceptor {
    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);
        asyncManager.registerCallableInterceptor("customAsyncException", new CallableProcessingInterceptor() {
            @Override
            public <T> Object handleError(NativeWebRequest request, Callable<T> task, Throwable t) throws Exception {
                if (t instanceof IOException && t.getMessage().contains("Broken pipe")) {
                    // 客户端断开，直接返回null，不触发错误分发
                    return null;
                }
                throw t;
            }
        });
        return true;
    }
}

然后在Spring配置中注册这个拦截器：

@Configuration
public class WebConfig implements WebMvcConfigurer {
    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new AsyncExceptionInterceptor());
    }
}

这个方案会全局拦截异步请求中的异常，针对Broken Pipe直接忽略，避免Tomcat生成500错误。

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