RabbitListener 不会接收使用 AsyncRabbitTemplate 发送的每条消息

Question

我在 Spring-Boot Version 1.5.4 上使用 Spring-Boot 项目，具有 spring-boot-starter-amqp、spring-boot-starter-web-services 和 spring-ws-support v。 2.4.0。到目前为止，我已经成功创建了一个 @RabbitListener 组件，当消息通过 rabbitTemplate.sendAndReceive(uri, message) 发送到代理时，它完全按照预期执行。我试着看看如果为此使用 AsyncRabbitTemplate 会发生什么，因为消息处理可能需要一段时间，而且我不想在等待响应时锁定我的应用程序。

问题是：我放入队列的第一条消息甚至没有被侦听器接收到。回调仅使用已发布的消息而不是返回的消息来确认成功。

听众：

@RabbitListener(queues = KEY_MESSAGING_QUEUE)
public Message processMessage(@Payload byte[] payload, @Headers Map<String, Object> headers) {
    try {
        byte[] resultBody = messageProcessor.processMessage(payload, headers);
        MessageBuilder builder = MessageBuilder.withBody(resultBody);
        if (resultBody.length == 0) {
            builder.setHeader(HEADER_NAME_ERROR_MESSAGE, "Error occurred during processing.");
        }
        return builder.build();
    } catch (Exception ex) {
        return MessageBuilder.withBody(EMPTY_BODY)
                .setHeader(HEADER_NAME_ERROR_MESSAGE, ex.getMessage())
                .setHeader(HEADER_NAME_STACK_TRACE, ex.getStackTrace())
                .build();
    }
}

当我执行测试时，一个测试失败，而第二个测试成功。 class 被注释为 @RunWith(SpringJUnit4ClassRunner.class) 和 @SpringBootTest(classes = { Application.class, Test.TestConfiguration.class }) 并且 @ClassRule 为 BrokerRunning.isRunningWintEmptyQueues(QUEUE_NAME)

测试配置（内部class）：

public static class TestConfiguration {

    @Bean // referenced in the tests as art
    public AsyncRabbitTemplate asyncRabbitTemplate(ConnectionFactory connectionFactory, RabbitTemplate rabbitTemplate) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueueNames(QUEUE_NAME);
        return new AsyncRabbitTemplate(rabbitTemplate, container);
    }

    @Bean
    public MessageListener messageListener() {
        return new MessageListener();
    }
}

测试：

@Test
public void shouldListenAndReplyToQueue() throws Exception {
    doReturn(RESULT_BODY)
            .when(innerMock)
            .processMessage(any(byte[].class), anyMapOf(String.class, Object.class));
    Message msg = MessageBuilder
            .withBody(MESSAGE_BODY)
            .setHeader("header", "value")
            .setHeader("auth", "entication")
            .build();

    RabbitMessageFuture pendingReply = art.sendAndReceive(QUEUE_NAME, msg);
    pendingReply.addCallback(new ListenableFutureCallback<Message>() {

        @Override
        public void onSuccess(Message result) { }

        @Override
        public void onFailure(Throwable ex) {
            throw new RuntimeException(ex);
        }
    });

    while (!pendingReply.isDone()) {}
    result = pendingReply.get();
    // assertions omitted
}

测试 2：

@Test
public void shouldReturnExceptionToCaller() throws Exception {
    doThrow(new SSLSenderInstantiationException("I am a message", new Exception()))
            .when(innerMock)
            .processMessage(any(byte[].class), anyMapOf(String.class, Object.class));
    Message msg = MessageBuilder
            .withBody(MESSAGE_BODY)
            .setHeader("header", "value")
            .setHeader("auth", "entication")
            .build();

    RabbitMessageFuture pendingReply = art.sendAndReceive(QUEUE_NAME, msg);
    pendingReply.addCallback(/*same as above*/);

    while (!pendingReply.isDone()) {}
    result = pendingReply.get();
    //assertions omitted
}

当我运行两个一起测试时，第一个执行的测试失败，而第二个调用成功。当我运行分别测试时，都失败了。当我添加一个 @Before-Method，它使用 AsyncRabbitTemplate art 将任何消息放入队列时，两个测试都可能通过，或者第二个测试可能不通过，所以除了意外之外，行为也不一致。

有趣的是，传递给方法的回调在调用侦听器之前报告成功，并将发送的消息报告为结果。

这里唯一缺少的 class 是通用配置 class，它用 @EnableRabbit 注释并具有以下内容：

@Bean
public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setConcurrentConsumers(10);
    return factory;
}

我尝试过的其他东西：

自己专门创建AsyncRabbitTemplate，每次消息处理前后手动启动和停止-> 两个测试都成功
增加/减少接收超时 -> 无效
删除并更改回调 -> 无效
通过注入再次显式创建队列 RabbitAdmin -> 无效
将回调提取到常量 -> 测试甚至没有正确启动
如上所述，我直接使用 RabbitTemplate，效果完全符合预期

如果有人知道缺少什么，我会很高兴听到。

Answer 1

您不能对请求和回复使用同一个队列...

@Bean // referenced in the tests as art
public AsyncRabbitTemplate asyncRabbitTemplate(ConnectionFactory connectionFactory, RabbitTemplate rabbitTemplate) {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
    container.setQueueNames(QUEUE_NAME);
    return new AsyncRabbitTemplate(rabbitTemplate, container);
}

将收听 QUEUE_NAME 上的回复，所以...

RabbitMessageFuture pendingReply = art.sendAndReceive(QUEUE_NAME, msg);

...简单地向自己发送消息。看起来你打算...

RabbitMessageFuture pendingReply = art.sendAndReceive(KEY_MESSAGING_QUEUE, msg);

RabbitListener 不会接收使用 AsyncRabbitTemplate 发送的每条消息

RabbitListener does not pick up every message sent with AsyncRabbitTemplate

java

spring

integration-testing

asynchronous

spring-rabbit