Spring Cloud Stream - 如何处理下游块？

Question

在我们的 Kafka 集群计划停机期间，我们基本上遇到了以下问题 How to specify timeout for sending message to RabbitMQ using Spring Cloud Stream?（显然是使用 Kafka 而不是 RabbitMQ）。

@GaryRussell 的回答：

频道sendTimeout 仅适用于频道本身可以阻塞的情况，例如带有当前已满的有界队列的QueueChannel；调用者将阻塞，直到队列中有空间可用，或者发生超时。

在这种情况下，阻塞在通道的下游，因此 sendTimeout 无关紧要（无论如何，它是一个 DirectChannel，无论如何都不能阻塞，订阅的处理程序直接在调用线程上调用）。

您看到的实际阻塞很可能在rabbitmq客户端中的socket.write()中，它没有超时且不可中断；调用线程无法执行任何操作来“超时”写入。

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

很好地解释了为什么有问题的方法 (org.springframework.integration.channel.AbstractSubscribableChannel#doSend) 没有考虑到timeout。但是，这对我来说仍然有点奇怪。

在spring-integration-kafka-3.2.1.RELEASE-sources.jar!/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java:566 中，我们可以看到，如果需要sync 行为：

565    if (this.sync) {
566        Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class);
567        if (sendTimeout == null || sendTimeout < 0) {
568            future.get();
569        }
570        else {
571            try {
572                future.get(sendTimeout, TimeUnit.MILLISECONDS);
573            }
574            catch (TimeoutException te) {
575                throw new MessageTimeoutException(message, "Timeout waiting for response from KafkaProducer", te);
576            }
577        }
578    }

被调用，其中考虑了超时。 sendTimeoutExpression 被分配了一个默认值：

    private static final long DEFAULT_SEND_TIMEOUT = 10000;
    private Expression sendTimeoutExpression = new ValueExpression<>(DEFAULT_SEND_TIMEOUT);

然而，我们的堆栈跟踪揭示了一些不同的东西：

"pool-1-thread-3" - Thread t@108
   java.lang.Thread.State: TIMED_WAITING
    at sun.misc.Unsafe.park(Native Method)
    - parking to wait for <4ebda621> (a org.springframework.util.concurrent.SettableListenableFuture$SettableTask)
    at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
    at java.util.concurrent.FutureTask.awaitDone(FutureTask.java:426)
    at java.util.concurrent.FutureTask.get(FutureTask.java:204)
    at org.springframework.util.concurrent.SettableListenableFuture.get(SettableListenableFuture.java:134)
*   at org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler.processSendResult(KafkaProducerMessageHandler.java:572)
    at org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler.handleRequestMessage(KafkaProducerMessageHandler.java:414)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:134)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:69)
    at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder$SendingHandler.handleMessageInternal(AbstractMessageChannelBinder.java:1035)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:69)
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:570)

标有* 的调用对应于future.get(sendTimeout, TimeUnit.MILLISECONDS); 调用。

看到底层客户端似乎支持它（鉴于future.get() 调用支持超时这一事实），如何设置？我可以在活页夹引用中找到的唯一两个属性（参见 here）是 spring.cloud.stream.kafka.binder.healthTimeout 和 batchTimeout，据我所知，这不会影响此设置。

看看KafkaProducerMessageHandler是如何在私有类org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.ProducerConfigurationMessageHandler中构造的，bean覆盖似乎不是推荐的方式。

Answer 1

似乎没有记录，但类似于侦听器容器定制器https://docs.spring.io/spring-cloud-stream/docs/3.1.2/reference/html/spring-cloud-stream.html#_advanced_consumer_configuration，您可以添加ProducerMessageHandlerCustomizer @Bean 以在消息处理程序上设置任意属性。

在较新版本的处理程序中，超时始终配置为至少与ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG 一样多，以避免误报（在处理程序超时后发布成功）。