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

Question

我正在 Spring-Boot 版本 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();
    }
}

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

TestConfiguration(内部类):

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时-方法，使用 AsyncRabbitTemplate art将任何消息放入队列中，两个测试都可能通过，或者第二个测试可能不通过，因此除了意外之外，行为也不一致。

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

唯一缺少的类是通用配置类，它用 @EnableRabbit 注释。并具有以下内容:

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

我尝试过的其他事情:

• 专门创建AsyncRabbitTemplate我自己，在每个消息处理之前和之后手动启动和停止它 -> 两个测试都成功
• 增加/减少接收超时 -> 无效果
• 删除并更改回调 -> 无效
• 使用注入(inject)的 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);