我们在发送消息的过程中遇到了网络问题,这导致所有线程都处于阻塞状态。我们使用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)
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参数被忽略了,我如何配置它来避免长时间的阻塞状态吗?
谢谢你,谢谢
2条答案
按热度按时间watbbzwu1#
通道
sendTimeout
仅在通道本身可以阻塞时适用,例如QueueChannel
,有界队列当前已满;呼叫者将阻塞,直到队列中的空间变得可用或超时发生。在这种情况下,块位于通道的下游,因此sendTimeout是不相关的(在任何情况下,它都是一个DirectChannel,无论如何都不能阻塞,订阅的处理程序直接在调用线程上调用)。
您看到的实际阻塞最有可能是在rabbitmq客户端的
socket.write()
中,它没有超时,并且不可中断;调用线程不能做任何事情来“超时”写入。我所知道的唯一可能的解决方案是通过调用连接工厂上的
resetConnection()
来强制关闭rabbit连接。omhiaaxx2#
保护向RabbitMQ发送消息的代码不被阻塞的一种方法是将代码 Package 在线程中并给予它一个执行超时。例如,
ExecutorService
类提供了一种在调用可调用对象时指定超时的方法