Spring Integration DSL错误处理器如何获取原处理线程ID

你当前配置中errorChannel绑定在轮询器（Poller）上，业务处理逻辑运行在你自定义的5个线程的executor线程池中，一旦业务线程抛出异常，异常会被封装为ErrorMessage提交给Poller的调度线程处理，因此错误处理逻辑运行在Poller调度线程上，和原业务处理线程不属于同一个线程，直接在错误处理中获取当前线程ID无法拿到业务线程的ID。

以下三种方案均可满足需求，你可以根据业务场景选择：

方案1：业务节点前置捕获异常，透传原线程ID到错误通道 #

不需要调整现有错误通道的绑定逻辑，直接在业务处理节点（transform/handle）外层包try-catch，捕获异常时直接获取当前线程ID，存入消息头后转发到错误通道。
修改后的代码示例：

@Bean
public IntegrationFlow testFile() {
    IntegrationFlowBuilder testChannel = IntegrationFlows.from(Files.inboundAdapter(new File("d:/input-files/")),
                    e -> e.poller(Pollers.fixedDelay(5000L).maxMessagesPerPoll(5)
                            .errorChannel("testChannel")))
            .channel(MessageChannels.executor(Executors.newFixedThreadPool(5)))
            .transform(o -> {
                long originalThreadId = Thread.currentThread().getId();
                try {
                    // 你的业务逻辑
                    throw new RuntimeException("Failing on purpose");
                } catch (Exception e) {
                    // 封装异常和原线程ID到消息头
                    return MessageBuilder.withPayload(o)
                            .setHeader("originalThreadId", originalThreadId)
                            .setHeader("businessError", e)
                            .build();
                }
            })
            // 路由异常消息到错误处理通道，正常消息走后续业务流程
            .routeToRecipients(route -> route
                    .recipient("testChannel", m -> m.getHeaders().containsKey("businessError"))
                    .recipient("normalProcessChannel", m -> !m.getHeaders().containsKey("businessError"))
            );
    return testChannel.get();
}

@Bean
public StandardIntegrationFlow errorChannelHandler() {
    return IntegrationFlows.from("testChannel")
            .handle(message -> {
                Long originalThreadId = (Long) message.getHeaders().get("originalThreadId");
                Exception error = (Exception) message.getHeaders().get("businessError");
                log.info("Handling error, original business thread id: {}, error info: {}", originalThreadId, error);
            }).get();
}

方案2：自定义业务异常类封装线程ID #

如果不想调整现有消息路由逻辑，可以自定义带线程ID属性的业务异常类，业务抛出异常时直接把当前线程ID存入异常属性，错误处理时从异常栈中取出即可。
代码示例：

// 自定义业务异常类
public class BusinessProcessException extends RuntimeException{
    private final long originalThreadId;

    public BusinessProcessException(String message, long originalThreadId) {
        super(message);
        this.originalThreadId = originalThreadId;
    }

    public long getOriginalThreadId() {
        return originalThreadId;
    }
}

// 业务节点抛出异常时传入线程ID
.transform(o -> {
    throw new BusinessProcessException("Failing on purpose", Thread.currentThread().getId());
})

// 错误处理中取线程ID
@Bean
public StandardIntegrationFlow errorChannelHandler() {
    return IntegrationFlows.from("testChannel")
            .handle(message -> {
                MessagingException exception = (MessagingException) message.getPayload();
                if (exception.getCause() instanceof BusinessProcessException) {
                    BusinessProcessException businessEx = (BusinessProcessException) exception.getCause();
                    long originalThreadId = businessEx.getOriginalThreadId();
                    log.info("Handling error, original business thread id: {}", originalThreadId);
                }
            }).get();
}

方案3：通过MDC透传线程上下文 #

如果需要同时传递更多线程上下文信息（不止线程ID），可以使用SLF4J的MDC工具，业务处理前将线程ID存入MDC，自定义线程池的TaskDecorator实现MDC上下文透传，错误处理时直接从MDC中取值即可。

如果需要实现系统关闭前等待所有业务线程执行完成，建议替换Executors.newFixedThreadPool(5)为Spring提供的ThreadPoolTaskExecutor，该类原生提供getActiveCount()方法可以获取当前活跃线程数，系统关闭钩子中循环等待活跃数降为0即可。

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