使用AsyncRabbitTemplate时回复超时：：SendandReceive-RabbitMQ

Question

我最近在Spring Boot应用程序中从使用标准Rabbit模板改为使用异步Rabbit模板。在此过程中，我从标准的send方法切换到使用sendandreceive方法。

进行此更改似乎不会影响向RabbitMQ发布消息，但是现在在发送消息时，我确实看到了如下所示的堆栈跟踪：

org.springframework.amqp.core.AmqpReplyTimeoutException: Reply timed out
    at org.springframework.amqp.rabbit.AsyncRabbitTemplate$RabbitFuture$TimeoutTask.run(AsyncRabbitTemplate.java:762) [spring-rabbit-2.3.10.jar!/:2.3.10]
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54) [spring-context-5.3.9.jar!/:5.3.9]

public static void main(String[] args) {
        com.rabbitmq.client.ConnectionFactory cf = new com.rabbitmq.client.ConnectionFactory();
        cf.setHost("localhost");
        cf.setPort(5672);
        cf.setUsername("<my-username>");
        cf.setPassword("<my-password>");
        cf.setVirtualHost("<my-vhost>");

        ConnectionFactory connectionFactory = new CachingConnectionFactory(cf);

        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setExchange("primary");
        rabbitTemplate.setUseDirectReplyToContainer(true);
        rabbitTemplate.setReceiveTimeout(10000);
        rabbitTemplate.setReplyTimeout(10000);
        rabbitTemplate.setUseChannelForCorrelation(true);

        AsyncRabbitTemplate asyncRabbitTemplate = new AsyncRabbitTemplate(rabbitTemplate);
        asyncRabbitTemplate.start();

        System.out.printf("Async Rabbit Template Running? %b\n", asyncRabbitTemplate.isRunning());

        MessageBuilderSupport<MessageProperties> props = MessagePropertiesBuilder.newInstance()
            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
            .setMessageId(UUID.randomUUID().toString())
            .setHeader(PUBLISH_TIME_HEADER, Instant.now(Clock.systemUTC()).toEpochMilli())
            .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);

        asyncRabbitTemplate.sendAndReceive(
            "1.1.1.csv-routing-key",
            new Message(
                "a,test,csv".getBytes(StandardCharsets.UTF_8),
                props.build()
            )
        ).addCallback(new ListenableFutureCallback<>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.printf("Error sending message:\n%s\n", ex.getLocalizedMessage());
            }

            @Override
            public void onSuccess(Message result) {
                System.out.println("Message successfully sent");
            }
        });
    }

我确信我只是缺少一个配置选项，但任何帮助都将是徒劳的。

谢谢。：）

Answer 1

设置回复超时到一些更大的数字，看看效果。

    rabbitTemplate.setReplyTimeout(60000);

https://docs.spring.io/spring-amqp/reference/html/#reply-timeout