我试图了解LoadbalanceRSocketClient SpringBoot 应用程序 ( RSocketRequester)上下文中的正确配置和使用模式。

我有两个 RSocket 服务器后端（SpringBoot、RSocket 消息传递）RSocketRequester在客户端上运行和配置，如下所示：

List<LoadbalanceTarget> servers = new ArrayList<>();
for (String url: backendUrls) {
  HttpClient httpClient = HttpClient.create()
    .baseUrl(url)
    .secure(ssl -> 
       ssl.sslContext(SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE)));
  servers.add(LoadbalanceTarget.from(url, WebsocketClientTransport.create(httpClient, url)));
}

// RSocketRequester.Builder is autowired by Spring boot
RSocketRequester requester = builder
  .setupRoute("/connect")
  .setupData("test")
  //.rsocketConnector(connector -> connector.reconnect(Retry.fixedDelay(60, Duration.ofSeconds(1))))
 .transports(Flux.just(servers), new RoundRobinLoadbalanceStrategy());   

配置完成后，请求者将在计时器循环中重复使用，如下所示：

@Scheduled(fixedDelay = 10000, initialDelay = 1000)
public void timer() {
  requester.route("/foo").data(Data).send().block();
}

它工作 - 客户端启动，连接到其中一台服务器并将消息推送到它。如果我终止客户端连接的服务器，客户端会在下一个计时器事件中重新连接到另一台服务器。如果我再次启动第一个服务器并杀死第二个服务器，客户端将不再连接，并且在客户端观察到以下异常：

java.util.concurrent.CancellationException: Pool is exhausted
    at io.rsocket.loadbalance.RSocketPool.select(RSocketPool.java:202) ~[rsocket-core-1.1.0.jar:na]
    at io.rsocket.loadbalance.LoadbalanceRSocketClient.lambda$fireAndForget$0(LoadbalanceRSocketClient.java:49) ~[rsocket-core-1.1.0.jar:na]
    at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:125) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:220) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1784) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.MonoZip$ZipCoordinator.signal(MonoZip.java:251) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.MonoZip$ZipInner.onNext(MonoZip.java:336) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1784) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.MonoCallable.subscribe(MonoCallable.java:61) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.Mono.subscribe(Mono.java:3987) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.Mono.subscribe(Mono.java:3987) ~[reactor-core-3.4.0.jar:3.4.0]
    at reactor.core.publisher.Mono.block(Mono.java:1678) ~[reactor-core-3.4.0.jar:3.4.0]

我怀疑我没有正确配置请求者或没有正确使用它。希望任何提示，因为文档和测试在这方面似乎非常薄弱。

理想情况下，我希望客户端在服务器/连接失败时透明地切换到任何下一个可用的服务器。现在重新连接尝试似乎只在下一次调用timer()方法时发生，这并不理想，因为客户端需要处理来自服务器的传入消息。我观察到的另一件事是，即使如此"/foo"，FnF 路由也是如此，除非我block()在send()服务器从未收到呼叫之后这样做。

回答

ReactiveDiscoveryClient discoveryClient = ...

Mono<List<LoadbalanceTarget>> serversMono = discoveryClient
    .getInstances(serviceGroupName)
    .map(si -> {
        HttpClient httpClient = HttpClient.create()
          .baseUrl(si.getUri())
          .secure(ssl -> ssl.sslContext(
              SslContextBuilder.forClient()
                         .trustManager(InsecureTrustManagerFactory.INSTANCE)
          ));
        return LoadbalanceTarget.from(si.getUri(), WebsocketClientTransport.create(httpClient, "/rsocket")));
    })
    .collectList()

// RSocketRequester.Builder is autowired by Spring boot
RSocketRequester requester = builder
  .setupRoute("/connect")
  .setupData("test")
  .transports(serversMono.flux(), new RoundRobinLoadbalanceStrategy());   

ReactiveDiscoveryClient discoveryClient = ...

Flux<List<LoadbalanceTarget>> serversFlux = discoveryClient
    .getInstances(serviceGroupName)
    .map(si -> {
        HttpClient httpClient = HttpClient.create()
          .baseUrl(si.getUri())
          .secure(ssl -> ssl.sslContext(
              SslContextBuilder.forClient()
                         .trustManager(InsecureTrustManagerFactory.INSTANCE)
          ));
        return LoadbalanceTarget.from(si.getUri(), WebsocketClientTransport.create(httpClient, "/rsocket")));
    })
    .collectList()
    .repeatWhen(f -> f.delayElements(Duration.ofSeconds(1))) // <- continuously retrieve new List of ServiceInstances

// RSocketRequester.Builder is autowired by Spring boot
RSocketRequester requester = builder
  .setupRoute("/connect")
  .setupData("test")
  .transports(servers, new RoundRobinLoadbalanceStrategy());