@多个类中同一队列的RabbitListener

Question

我想知道，在Spring AMQP中，是否可以根据负载类型在多个类中接收来自同一队列的消息。

我知道在类中使用@RabbitListener注释，然后将@RabbitHandler放在方法上，但我希望在保持单个队列的同时将消息处理的复杂性拆分为多个类。

当前使用的版本：Spring AMQP v2.0.3以及RabbitMQ。

Answer 1

或者，我发现，如果您有足够高版本的AQMP和Rabbit（mine=spring-rabit 2.1.3， spring-starter-aqmp 2.1.2），您可以执行以下多态：

@RabbitListener(id="multi", queues = "somequeuename")
public class SomeService 
{
    @RabbitHandler
    public void handleADTO(@Payload ADTO adto) {
         System.out.print(adto.url);
    }

    @RabbitHandler
    public void handleADTO2(@Payload ADTO2 adto) {
         System.out.print(adto.url);
    }
}

但是，开箱即用，这不起作用。您需要配置另一个“袋豆”：

@Bean
public SimpleRabbitListenerContainerFactory myFactory(
        SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory =
            new SimpleRabbitListenerContainerFactory();
    configurer.configure(factory, connectionFactory);
    return factory;
}

@Bean
Jackson2JsonMessageConverter jsonMessageConverter() {
    return new Jackson2JsonMessageConverter();
}

@Bean
RabbitTemplate rabbitTemplate(Jackson2JsonMessageConverter converter, ConnectionFactory connectionFactory) {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setMessageConverter(new Jackson2JsonMessageConverter());
    return template;
}

只需将其放入你的应用程序或其他类中。

另请参见

https://docs.spring.io/spring-amqp/docs/current/reference/html/

Answer 2

是的，您可以，但它采用了一种稍微不同的方法：您需要监听通用消息类型，进行一些切换，并进行自己的反序列化。当然，您可以将代码完全隐藏在某个地方（基类、注释…）

下面的例子可以扩展到任何其他类型。上面的示例是对A和B DTO类型进行过滤。

void receive(ADTO dto)
{
    System.out.print(dto.url);
}

void receive(BDTO dto)
{
    System.out.print(dto.url);
}
@RabbitListener(queues = "your.queue.name")
public void listenMesssage(Message message) 
 {
    try 
    {
        String typeId = message.getMessageProperties().getHeaders().get("__TypeId__").toString();
        String contentType = message.getMessageProperties().getContentType();
        if (contentType != "application/json" || typeId == null || !typeId.contains(ADTO.class.toString()))
        {
            //TODO log warning
            System.out.print("type not supported by this service");
            return;
        }
        Object receivedObject = new Jackson2JsonMessageConverter().fromMessage(message);

        if (receivedObject instanceof ADTO)
        {
            receive((ADTO)receivedObject);
            System.out.print("ADTO");
        }
        //else

或者，您也可以这样进行序列化：

....
String typeId = message.getMessageProperties().getHeaders().get("__TypeId__").toString();
byte[] binMsg =  message.getBody();
String strMsg  = new String(binMsg, StandardCharsets.UTF_8);
ObjectMapper mapper = new ObjectMapper();
if (typeId.contains("ADTO"))
{
   receive(mapper.readValue(strMsg, ADTO.class ));
}
else
...            

Answer 3

嗯，这是不可能的。那么你想要的方式就不会是排队了。设计一个监听器并根据负载类型分配给它的方法，这实际上是一个架构决策。

作为一种解决方法，我可以建议您将单个@RabbitListener类中的逻辑委托给这些业务服务：

    @RabbitListener(queues = "foo")
    public class MyListener {

        private final ServiceA serviceA;

        private final ServiceB serviceB;

        public MyListener(ServiceA serviceA, ServiceB serviceB) {
              this.serviceA = serviceA;
              this.serviceB = serviceB;
        }


        @RabbitHandler
        public void handleA(A a) {
             this.serviceA.handle(a);
        }

        @RabbitHandler
        public void handleB(B b) {
             this.serviceB.handle(b);
        }
    }