如何在 Reactor Netty 中处理 io.netty.channel.ConnectTimeoutException

Question

我正在尝试以反应方式使用 Reactor Netty TcpClient 与可能无法访问的主机进行交互。这是通道初始化逻辑的示例：

ConnectionProvider connectionProvider = ConnectionProvider.fixed("fixed", 50);
TcpClient.create(connectionProvider)
  .host(host).port(port)
  .wiretap(true)
  .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 50)
  .doOnConnect(x -> log.trace("Connect to {}:{}", host, port))
  .doOnConnected(conn -> log.trace("Connected {}", conn.channel()))
  .connect()
  .subscribe(this::utilizeConnection);

我收到的输出：

2019-09-04 08:23:13.612 TRACE 71988 --- [ioEventLoop-4-3] c.c.pcb.poc.network.tcp.NettyTcpSender   : Connect to 192.168.88.210:2000
2019-09-04 08:23:13.684  WARN 71988 --- [actor-tcp-nio-4] io.netty.util.concurrent.DefaultPromise  : An exception was thrown by reactor.netty.resources.PooledConnectionProvider$DisposableAcquire.operationComplete()

reactor.core.Exceptions$ErrorCallbackNotImplemented: io.netty.channel.ConnectTimeoutException: connection timed out: /192.168.88.210:2000
Caused by: io.netty.channel.ConnectTimeoutException: connection timed out: /192.168.88.210:2000
    at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.run(AbstractNioChannel.java:267) ~[netty-transport-4.1.36.Final.jar:4.1.36.Final]
    at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38) ~[netty-common-4.1.36.Final.jar:4.1.36.Final]
    at io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:127) ~[netty-common-4.1.36.Final.jar:4.1.36.Final]
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163) [netty-common-4.1.36.Final.jar:4.1.36.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:405) [netty-common-4.1.36.Final.jar:4.1.36.Final]
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500) [netty-transport-4.1.36.Final.jar:4.1.36.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor.run(SingleThreadEventExecutor.java:906) [netty-common-4.1.36.Final.jar:4.1.36.Final]
    at io.netty.util.internal.ThreadExecutorMap.run(ThreadExecutorMap.java:74) [netty-common-4.1.36.Final.jar:4.1.36.Final]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_181]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoCreate] :
    reactor.core.publisher.Mono.create(Mono.java:183)
    reactor.netty.resources.PooledConnectionProvider.acquire(PooledConnectionProvider.java:130)
Error has been observed by the following operator(s):
    |_  Mono.create ⇢ reactor.netty.resources.PooledConnectionProvider.acquire(PooledConnectionProvider.java:130)
    |_  Mono.doOnSubscribe ⇢ reactor.netty.tcp.TcpClientDoOn.connect(TcpClientDoOn.java:58)

'inbound' 和 'outbound' 有一个专门的方法来处理它们的错误，但它们在 Connection 实例之上工作，如果您获得 'connection timeout'.

我试过了：

我收到的异常包含在 'ErrorCallbackNotImplemented' 中。但是我找不到任何方法来实现任何 'ErrorCallback'
日志包含来自 'io.netty.util.concurrent.DefaultPromise' 的警告消息。但我无法找到一种方法来让自己的 Promise 以正确的方式处理它。
我发现没有任何可能以某种方式拦截连接超时的配置。
解决方法。创建连接的阻塞方法（.block() 而不是 .subscribe()）将允许我在普通的 try-catch 块中捕获任何创建异常的连接，但我将失去这种解决方法的反应方法的好处。

有人可以建议我至少一些东西来帮助我找到处理 'io.netty.channel.ConnectTimeoutException' 的正确方法吗？

Answer 1

不要忘记实施您的错误回调

通常 reactor.core.Exceptions$ErrorCallbackNotImplemented 发生在基于 labmda 的订阅 .subscribe 方法（Mono 和 Flux 相同）。

如果你要查看来源 here and here，你会找到抛出 reactor.core.Exceptions$ErrorCallbackNotImplemented 的地方！

行动点数

为了处理原始 io.netty.channel.ConnectTimeoutException 我建议查看原始 Project Reactor 文档的 Handling Errors 部分

如何在 Reactor Netty 中处理 io.netty.channel.ConnectTimeoutException

How to handle in Reactor Netty an io.netty.channel.ConnectTimeoutException

java

project-reactor

reactor-netty

不要忘记实施您的错误回调

行动点数