如何指定使用向RabbitMQ发送消息的超时？

Question

我们在发送消息的过程中遇到了网络问题，这导致所有线程处于阻塞状态。我们使用org.springframework.cloud:spring-cloud-stream:2.0.1.RELEASE和org.springframework:spring-messaging:5.0.8.RELEASE向RabbitMQ代理发送消息。绑定接口：

interface MessagingSource {
    @Output("bindingTargetName")
    fun messageChannelOutput(): MessageChannel
}

用法：

 val isSent = messageSource.messageChannelOutput().send(message)

在org.springframework.integration.channel.AbstractSubscribableChannel#doSend方法中，MessageChannel#send(Message，long)方法也有以毫秒为单位的超时作为第二个参数，但是它被忽略了：

@Override
protected boolean doSend(Message<?> message, long timeout) { // timeout is ignored in this method
    try {
        return getRequiredDispatcher().dispatch(message);
    }
    catch (MessageDispatchingException e) {
        String description = e.getMessage() + " for channel '" + this.getFullChannelName() + "'.";
        throw new MessageDeliveryException(message, description, e);
    }
}

您能否解释为什么忽略超时参数，以及如何配置它以避免长期阻塞状态？

谢谢!

Answer 1

只有当通道本身可以阻塞时，通道sendTimeout才会应用，例如，一个当前已满的有界队列的QueueChannel；调用方将阻塞，直到队列中的任何一个空间可用，或者超时发生。

在这种情况下，块位于通道的下游，因此sendTimeout是不相关的(无论如何，它是一个不能阻塞的DirectChannel，订阅的处理程序直接在调用线程上调用)。

您所看到的实际阻塞很可能发生在rabbitmq客户端的socket.write()中，该客户机没有超时，也不可中断；调用线程无法“超时”写入。

我知道的唯一可能的解决方案是通过调用连接工厂上的resetConnection()来强制关闭兔子连接。