如何在 Spring AMQP 中使用拦截器

Question

有没有办法在调用 template.convertAndSend 后，在消息传递到 RabbitMQ 之前拦截消息。

还有什么方法可以在到达处理程序之前拦截消息？

我可以使用 PostProcessor 为发布者处理消息，但更喜欢使用拦截器。

public class TestPostProcessor implements MessagePostProcessor {

    @Autowired
    Tracer defaultTracer;

    @Override
    public Message postProcessMessage(Message message) throws AmqpException {
        //.....
        //.... 
        return message;
    }
}

有什么建议吗？

Answer 1

MessagePostProcessor 是一种拦截器。

有两种方法可以调用其中一种 - 使用将 MPP 作为参数的重载 convertAndSend() 方法之一，或者使用 setBeforePublishPostProcessors() 将一个或多个添加到 RabbitTemplate。

您还可以使用setAfterReceivePostProcessors() 拦截收到的消息，在收到的消息从receive() 方法返回之前调用它。

侦听器容器还支持通过其setAfterReceivePostProcessors() 方法在接收之后和交付给侦听器之前的 MPP。

Answer 2

如果您想在 application.properties 文件中继续使用 spring boot 属性（来自 org.springframework.boot.autoconfigure.amqp.RabbitProperties），您可以提供自己的 RabbitListenerContainerFactory ：

  @Bean
  public CustomRabbitListenerContainerFactory rabbitListenerContainerFactory(
      SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory, MyContextMessageProcessor messageProcessor) {
    CustomRabbitListenerContainerFactory factory = new CustomRabbitListenerContainerFactory(messageProcessor);
    configurer.configure(factory, connectionFactory);
    return factory;
  }

CustomRabbitListenerContainerFactory.java：

public class CustomRabbitListenerContainerFactory
  extends org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory {

  private MessagePostProcessor[] messagePostProcessor;

  public CustomRabbitListenerContainerFactory(MessagePostProcessor... messagePostProcessor) {
    super();
    this.messagePostProcessor = messagePostProcessor;
  }

  @Override
  protected void initializeContainer(SimpleMessageListenerContainer instance, RabbitListenerEndpoint endpoint) {
    super.initializeContainer(instance, endpoint);
    instance.addAfterReceivePostProcessors(messagePostProcessor);
  }
}