Блог

Есть ли способ перехватывать сообщения после вызова template.convertAndSend, прежде чем сообщение будет доставлено в RabbitMQ.

Также есть какой-либо способ перехватить сообщение до того, как оно достигнет обработчика?

Я могу обработать сообщение с помощью PostProcessor для publisher, но предпочитаю использовать interceptor.

 public class TestPostProcessor implements MessagePostProcessor {

    @Autowired
    Tracer defaultTracer;

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

Есть предложения?

MessagePostProcessor это форма interceptor.

Есть два способа вызвать один из них — использовать один из перегруженных convertAndSend() методов, которые принимают MPP в качестве аргумента, или добавить один или несколько к RabbitTemplate using setBeforePublishPostProcessors() .

Вы также можете перехватывать полученные сообщения, используя setAfterReceivePostProcessors() который вызывается перед возвратом полученного сообщения из receive() метода.

Контейнер-прослушиватель также поддерживает MPPS после получения и перед доставкой слушателю с помощью своего setAfterReceivePostProcessors() метода.

Если вы хотите продолжать использовать свойства загрузки spring (из org.springframework.boot.autoconfigure.amqp.RabbitProperties) в вашем файле application.properties, вы можете предоставить свой собственный 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);
  }
}