java 如何使用Spring Cloud Stream指定向RabbitMQ发送消息的超时?

pkwftd7m  于 2023-09-29  发布在  Java
关注(0)|答案(2)|浏览(112)

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

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

使用方法:

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

MessageChannel#send(Message,long)方法的第二个参数也是timeout(毫秒),但在org.springframework.integration.channel.AbstractSubscribableChannel#doSend方法中被忽略了:

@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);
    }
}

你能解释一下为什么timeout参数被忽略了,我如何配置它来避免长时间的阻塞状态吗?
谢谢你,谢谢

watbbzwu

watbbzwu1#

通道sendTimeout仅在通道本身可以阻塞时适用,例如QueueChannel,有界队列当前已满;呼叫者将阻塞,直到队列中的空间变得可用或超时发生。
在这种情况下,块位于通道的下游,因此sendTimeout是不相关的(在任何情况下,它都是一个DirectChannel,无论如何都不能阻塞,订阅的处理程序直接在调用线程上调用)。
您看到的实际阻塞最有可能是在rabbitmq客户端的socket.write()中,它没有超时,并且不可中断;调用线程不能做任何事情来“超时”写入。
我所知道的唯一可能的解决方案是通过调用连接工厂上的resetConnection()来强制关闭rabbit连接。

omhiaaxx

omhiaaxx2#

保护向RabbitMQ发送消息的代码不被阻塞的一种方法是将代码 Package 在线程中并给予它一个执行超时。例如,ExecutorService类提供了一种在调用可调用对象时指定超时的方法

相关问题