如何从Advice中获取Spring AMQP Message？求场景适配示例

一、生产者端：将用户ID注入消息头 #

通过实现ProducerInterceptor拦截消息发送流程，把当前登录用户ID添加到消息自定义属性中：

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;

public class UserIdProducerInterceptor implements RabbitTemplate.ProducerInterceptor {

    private static final String USER_ID_HEADER = "X-User-Id";

    @Override
    public Message beforePublish(Message message, String exchange, String routingKey) {
        // 获取当前登录用户的认证信息
        Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
        if (authentication != null && authentication.isAuthenticated()) {
            // 根据实际业务获取用户ID，这里示例取用户名，可替换为用户ID字段
            String userId = authentication.getName(); 
            // 将用户ID写入消息头
            message.getMessageProperties().getHeaders().put(USER_ID_HEADER, userId);
        }
        return message;
    }

    // 以下方法为默认实现，无需额外修改
    @Override
    public void afterPublish(Message message, String exchange, String routingKey, boolean ack) {}

    @Override
    public void onReturnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {}

    @Override
    public RabbitTemplate setRabbitTemplate(RabbitTemplate rabbitTemplate) { return rabbitTemplate; }

    @Override
    public RabbitTemplate.ConfirmCallback getConfirmCallback() { return null; }

    @Override
    public RabbitTemplate.ReturnCallback getReturnCallback() { return null; }
}

将拦截器注册到RabbitTemplate：

@Configuration
public class RabbitProducerConfig {

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 添加生产者拦截器
        rabbitTemplate.addBeforePublishPostProcessors(new UserIdProducerInterceptor());
        return rabbitTemplate;
    }
}

二、消费者端：从消息头取出用户ID并注入SecurityContext #

实现MethodInterceptor拦截消费者方法调用，提取消息头中的用户ID，构建认证对象并设置到SecurityContextHolder：

import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.amqp.core.Message;
import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.util.StringUtils;

public class UserIdConsumerAdvice implements MethodInterceptor {

    private static final String USER_ID_HEADER = "X-User-Id";

    @Override
    public Object invoke(MethodInvocation invocation) throws Throwable {
        // 从方法参数中找到Message对象
        Message message = null;
        for (Object arg : invocation.getArguments()) {
            if (arg instanceof Message) {
                message = (Message) arg;
                break;
            }
        }

        if (message != null) {
            // 从消息头提取用户ID
            String userId = (String) message.getMessageProperties().getHeaders().get(USER_ID_HEADER);
            if (StringUtils.hasText(userId)) {
                // 根据业务构建认证对象，这里使用最简版，可添加权限信息
                Authentication auth = new UsernamePasswordAuthenticationToken(userId, null, null);
                SecurityContextHolder.getContext().setAuthentication(auth);
            }
        }

        try {
            // 执行消费者原业务方法
            return invocation.proceed();
        } finally {
            // 清理上下文，避免线程池复用导致的上下文污染
            SecurityContextHolder.clearContext();
        }
    }
}

将Advice注册到消费者容器工厂：

@Configuration
public class RabbitConsumerConfig {

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 添加消费者拦截Advice
        factory.setAdviceChain(new UserIdConsumerAdvice());
        return factory;
    }
}

关键说明 #

• 生产者端：beforePublish方法直接拿到待发送的Message对象，修改其消息头即可完成用户ID传递。
• 消费者端：通过遍历方法参数获取Message，提取用户ID后注入SecurityContext，最后必须清理上下文，防止线程复用引发的信息泄露。
• 消息头的键名（示例用X-User-Id）可根据业务调整，确保生产者和消费者保持一致。

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