Spring Integration：如何用Java配置为AmqpInboundChannelAdapter添加AroundAdvice？

好的，针对你的单线程同步流程需求，我们可以通过两种方式实现环绕增强来管理MDC字段：Spring Integration原生的RequestHandlerAdvice（更贴合场景，精准拦截消息处理逻辑），或者Spring AOP注解式切面（更灵活）。下面是具体的实现步骤：

方案一：使用Spring Integration RequestHandlerAdvice（推荐） #

这种方式直接作用于Spring Integration的消息处理器，比普通AOP更精准，不会拦截无关方法。

1. 自定义MDC增强类 #

继承AbstractRequestHandlerAdvice，实现环绕逻辑，用try-finally确保无论流程成功还是失败，MDC字段都会被清理：

import org.slf4j.MDC;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.messaging.Message;

public class MdcPopulatingAdvice extends AbstractRequestHandlerAdvice {

    @Override
    protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
        try {
            // 从消息中提取orderNr（根据你的实际消息结构调整逻辑）
            String orderNr = extractOrderNrFromMessage(message);
            // 设置MDC字段
            MDC.put("orderNr", orderNr);
            // 执行原有的消息处理流程（包括发送到WebServiceGateway）
            return callback.execute();
        } finally {
            // 强制清理MDC，避免残留影响后续日志
            MDC.remove("orderNr");
        }
    }

    private String extractOrderNrFromMessage(Message<?> message) {
        // 示例：从消息头获取orderNr，也可以从payload中提取
        return (String) message.getHeaders().get("orderNr", "default-order-id");
    }
}

2. 在Java配置中绑定到AmqpInboundChannelAdapter #

将自定义Advice加入到适配器的requestHandlerAdviceChain中，确保它包裹消息处理的全流程：

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.messaging.MessageChannel;
import java.util.List;

@Configuration
@EnableIntegration
public class AmqpIntegrationConfig {

    // 定义接收AMQP消息的通道（单线程同步流程用DirectChannel即可）
    @Bean
    public MessageChannel amqpInputChannel() {
        return new DirectChannel();
    }

    // 配置监听的AMQP队列
    @Bean
    public Queue inputQueue() {
        return new Queue("your-queue-name");
    }

    // 配置消息监听容器（单线程，设置concurrentConsumers=1）
    @Bean
    public SimpleMessageListenerContainer messageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(rabbitConnectionFactory()); // 假设你已配置RabbitConnectionFactory
        container.setQueues(inputQueue());
        container.setConcurrentConsumers(1);
        return container;
    }

    // 配置AmqpInboundChannelAdapter并绑定MDC增强
    @Bean
    public AmqpInboundChannelAdapter amqpInboundChannelAdapter() {
        AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(messageListenerContainer());
        adapter.setOutputChannel(amqpInputChannel());
        // 添加自定义MDC增强到处理器链
        adapter.setRequestHandlerAdviceChain(List.of(mdcPopulatingAdvice()));
        return adapter;
    }

    // 注册MDC增强Bean
    @Bean
    public MdcPopulatingAdvice mdcPopulatingAdvice() {
        return new MdcPopulatingAdvice();
    }

    // 配置WebServiceGateway相关的处理器（示例）
    @Bean
    public MessageHandler webServiceOutboundHandler() {
        // 这里替换为你的WebServiceOutboundGateway或自定义网关实现
        return new WebServiceOutboundGateway("http://your-target-webservice-url");
    }

    // 构建集成流：AMQP输入通道 -> WebService网关
    @Bean
    public IntegrationFlow amqpToWebServiceFlow() {
        return IntegrationFlows.from(amqpInputChannel())
                .handle(webServiceOutboundHandler())
                .get();
    }
}

方案二：使用Spring AOP注解式切面 #

如果你需要更灵活地拦截多个方法（比如同时增强Amqp适配器和WebService网关），可以用注解式切面：

1. 定义MDC切面类 #

import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.slf4j.MDC;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

@Aspect
@Component
public class MdcLoggingAspect {

    // 切入点：匹配Amqp适配器的handleMessage方法和WebService网关的处理方法
    @Around("execution(* org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.handleMessage(..)) || " +
            "execution(* com.yourpackage.YourWebServiceGateway.*(..))")
    public Object populateAndCleanMdc(ProceedingJoinPoint joinPoint) throws Throwable {
        try {
            // 从方法参数中获取Message对象
            Object[] args = joinPoint.getArgs();
            if (args.length > 0 && args[0] instanceof Message) {
                Message<?> message = (Message<?>) args[0];
                String orderNr = (String) message.getHeaders().get("orderNr", "default-order-id");
                MDC.put("orderNr", orderNr);
            }
            // 执行原方法逻辑
            return joinPoint.proceed();
        } finally {
            // 清理MDC字段
            MDC.remove("orderNr");
        }
    }
}

2. 开启AOP支持 #

在配置类上添加@EnableAspectJAutoProxy注解即可生效。

关键注意事项 #

• 必须用try-finally：无论流程中是否抛出异常，都要确保MDC字段被清理，避免线程复用（即使是单线程，后续消息也会用同一个线程）时残留旧的MDC值，导致日志混乱。
• 单线程的安全性：因为你的流程是单线程运行的，MDC的线程绑定特性完全适配，不会有线程安全问题。
• 消息提取逻辑：根据你的实际消息结构调整extractOrderNrFromMessage方法，比如从payload的业务对象中获取，而不是消息头。

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