java - spring amqp注解驱动一个队列两个监听器区分路由key

Question

实际上我没有运行这个。也许我误解了一些东西，无论如何这是不可能的。我正在尝试在同一个队列、相同的交换器上配置 2 个监听器，但只有路由键应该不同。我的问题是事情不知何故变得困惑。结果是监听器 A 收到了监听器 B 的消息。但只是有时，有时一切正常。有什么建议吗？

我的配置

@Bean
public ConnectionFactory connectionFactory() {
    CachingConnectionFactory connectionFactory = new CachingConnectionFactory(getHostname());
    connectionFactory.setUsername(getUsername());
    connectionFactory.setPassword(getPassword());
    return connectionFactory;
}

@Bean
public RabbitAdmin rabbitAdmin() {
    return new RabbitAdmin(connectionFactory());
}

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setMessageConverter(new CustomMessageConverter());
    factory.setConnectionFactory(connectionFactory());
    factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
    factory.setConcurrentConsumers(10);
    factory.setMaxConcurrentConsumers(10);
    return factory;
}

@Override
public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
    registrar.setMessageHandlerMethodFactory(myHandlerMethodFactory());
}

@Bean
public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
    DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
    factory.setMessageConverter(new MappingJackson2MessageConverter());
    return factory;
}

我的听众 A

@RabbitListener(bindings = @QueueBinding(value = @Queue(value = QUEUE, durable = "true"), exchange = @Exchange(value = EXCHANGE, type = "topic", durable = "true", ignoreDeclarationExceptions = "true"), key = "routingKeyA"))
public String myListenerA(@Payload PayloadA payload, @Header(AmqpHeaders.CORRELATION_ID) byte[] correlationId) { 

    return SUCCESS_RESPONSE;
}

我的监听器B

@RabbitListener(bindings = @QueueBinding(value = @Queue(value = QUEUE, durable = "true"), exchange = @Exchange(value = EXCHANGE, type = "topic", durable = "true", ignoreDeclarationExceptions = "true"), key = "routingKeyB"))
public String myListenerB(@Payload PayloadB payload, @Header(AmqpHeaders.CORRELATION_ID) byte[] correlationId) { 

    return SUCCESS_RESPONSE;
}

其他信息:此队列中有 20 个消费者。提前谢谢!

Answer 1

RabbitMQ 不是这样工作的；与 JMS 不同，无法从队列中选择消息(例如，基于路由键)。

您所做的就是将同一个队列绑定(bind)到具有 2 个不同路由键的交换机。所以，是的，任一监听器都会收到消息，无论消息如何到达队列。

使用 RabbitMQ，您需要为每个监听器一个单独的队列。当生产者发布到交换器时，代理将根据他使用的路由 key 将消息路由到正确的队列。

如果每个监听器有多个实例，消息将相应地分发(每个队列仅传递一次)。