如何在Spring Boot的Apache Camel项目中为所有SQS路由配置全局拦截器

很高兴帮你解决这个问题！在Apache Camel里实现全局SQS消息拦截其实很简单，我们可以利用Camel提供的全局拦截器和OnCompletion机制，实现和你之前Spring Cloud AWS类似的效果，不用在每个路由里重复写逻辑。下面给你最贴合需求的实现方案：

方案：全局拦截+统一清理（完美匹配你的MDC场景） #

这个方案分为两步：一是拦截所有SQS入站消息并设置MDC，二是在消息处理完成（无论成功/失败）后统一清理MDC，避免线程池残留上下文。

1. 创建全局拦截处理器 #

先写一个负责MDC初始化的处理器，逻辑和你之前的handleMessage前置部分一致：

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.slf4j.MDC;
import org.springframework.stereotype.Component;
import java.util.UUID;

@Component
public class GlobalSqsMdcInitializer implements Processor {
    @Override
    public void process(Exchange exchange) throws Exception {
        // 这里可以根据实际需求从SQS消息头里获取真实RequestId，或者生成随机值
        String requestId = "RequestId=" + UUID.randomUUID();
        MDC.put("requestId", requestId);
        // 可选：把RequestId存入Exchange，方便后续处理器使用
        exchange.setProperty("sqsRequestId", requestId);
    }
}

2. 创建全局MDC清理器 #

用Camel的OnCompletion接口实现消息处理后的MDC清理，确保无论成功失败都执行：

import org.apache.camel.Exchange;
import org.apache.camel.OnCompletion;
import org.slf4j.MDC;
import org.springframework.stereotype.Component;

@Component
public class GlobalMdcCleanup implements OnCompletion {
    @Override
    public void onComplete(Exchange exchange) {
        cleanUpMdc();
    }

    @Override
    public void onFailure(Exchange exchange) {
        cleanUpMdc();
    }

    private void cleanUpMdc() {
        // 建议只移除我们设置的key，避免影响其他MDC上下文
        MDC.remove("requestId");
        // 如果需要完全清空可以用MDC.clear()，但要谨慎使用
    }
}

3. 配置全局拦截规则 #

创建一个全局的RouteBuilder，配置拦截所有SQS入站消息，并绑定全局清理逻辑：

import org.apache.camel.builder.RouteBuilder;
import org.springframework.stereotype.Component;

@Component
public class GlobalSqsInterceptorRoute extends RouteBuilder {
    private final GlobalSqsMdcInitializer mdcInitializer;
    private final GlobalMdcCleanup mdcCleanup;

    // 构造注入依赖
    public GlobalSqsInterceptorRoute(GlobalSqsMdcInitializer mdcInitializer, GlobalMdcCleanup mdcCleanup) {
        this.mdcInitializer = mdcInitializer;
        this.mdcCleanup = mdcCleanup;
    }

    @Override
    public void configure() throws Exception {
        // 拦截所有aws2Sqs类型的入站端点（匹配你所有的SQS路由）
        interceptFrom("aws2Sqs:*")
                .process(mdcInitializer);

        // 全局绑定清理逻辑：所有路由处理完成后执行
        onCompletion()
                .onCompleteOnly(false) // 成功/失败都触发
                .bean(mdcCleanup);
    }
}

4. 原有路由无需修改 #

你的MessageRoute1和MessageRoute2完全不需要改动，全局拦截器会自动对所有SQS路由生效：消息会先经过GlobalSqsMdcInitializer设置MDC，处理完成后自动触发GlobalMdcCleanup清理。

关键说明 #

• interceptFrom("aws2Sqs:*")：用通配符匹配所有AWS SQS类型的入站端点，确保不会拦截其他类型的路由。
• OnCompletion：Camel的这个机制能保证消息处理全生命周期结束后执行清理，完美适配异步线程池场景，避免了try-finally可能提前执行的问题。
• 如果你需要从SQS消息本身获取真实的RequestId，可以从Exchange的消息头中读取：exchange.getIn().getHeader("RequestId", String.class)，替换掉随机生成的逻辑即可。

备注：内容来源于stack exchange，提问作者Samet Baskıcı