3

我试图了解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()服务器从未收到呼叫之后这样做。

4

3 回答 3

7

不断更新端点列表

LoadbalanceClient旨在与负责保持 a Listof aliveInstance的 Discovery 服务集成。也就是说，如果其中一项服务从集群中消失，则 Discovery 服务会更新其List可用Instance的 s。

另一方面，要实现客户端负载平衡，我们必须知道集群中可用服务的列表。很明显，要设置负载平衡，我们可以检索服务列表并将其提供给 Loadbalancer API。

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());

然而，想象一下我们处于一个完全分布式的环境中，现在每一个消失又出现的服务都运行在全新的主机和端口上（例如，kubernates 集群不固定在特定的 IP 地址上）。也就是说，负载平衡必须考虑这种情况并避免池中的死节点，它会从池中完全删除不健康的节点。

现在，如果所有节点都消失并在一段时间后出现，它们就不再包含在池中（并且如果Flux提供更新的 已完成，实际上，池已用尽，因为不会有新的更新来自Flux<List<LodbalanceTarget>>）。

但是，节点将自己注册到 Discovery 服务并可供观察。综上所述，我们必须定期从发现服务中提取信息以保持最新并不断更新池状态

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());

通过这样的设置，RSocketPool如果所有节点都从集群中消失，则不会耗尽，因为Flux<List<LoadbalanceTraget>>尚未完成并且最终可能会提供新的更新。

请注意，该实现足够智能，可以在发现服务的每次更新中保持活动节点。也就是说，如果池中有这样的服务实例，您将不会同时获得 2 个连接。

关于重新连接功能的旁注

您可能会注意到，它RSocketConnector提供了一个名为.reconnect. 乍一看，使用 似乎reconnect可以使您的连接保持正常运行并无限运行。不幸的是，事实并非如此。该.reconnect功能旨在Mono<RSocket>通过缓存语义保持您的可重用性，这意味着您可以在不同的地方多次创建@Bean Mono<RSocket> ...并自动装配它，而不必担心每个. 另一方面，如果给定断开连接（例如失去连接的情况），下一次订阅这样的 a将只针对所有并发呼叫抵抗一次新的订阅。subscribeRSocket instanceMono<RSocket>.subscribe.reconnectRSocketMono<RSocket>RSocket.subscribe

虽然这听起来很有用，RSocketPool但我们并不太依赖它，Mono<RSocket>只使用一次来解析和缓存 RSocketPool 中的 RSocket 实例。也就是说，如果此类 RSocket 将断开连接，我们将不会尝试Mono<RSocket>再次订阅给定的（我们假设设置的主机和端口将被更改）

于 2020-12-12T18:36:00.973 回答
1

关于 FnF 的问题，这是 Rx 模型的一部分。没有订阅，事件就不会发生。在订阅之前，您可以自由调用返回 Mono 而没有副作用的 API，任何其他行为都是错误。

  /**
   * Perform a Fire-and-Forget interaction via {@link RSocket#fireAndForget(Payload)}. Allows
   * multiple subscriptions and performs a request per subscriber.
   */
  Mono<Void> fireAndForget(Mono<Payload> payloadMono);

如果您调用此方法一次，然后订阅 3 次结果，它将执行 3 次。

于 2020-12-12T13:04:18.267 回答
0

奥莱，我尝试了你的建议，它在一定程度上有效，但我仍然不能完全得到我需要的行为。

我想做的是：

  • 客户端一次连接到一个（随机）后端
  • 如果后端或与后端的连接失败，客户端应尝试连接到下一个可用的后端。

我想我不能使用RoundRobinLoadbalanceStrategy它，因为它将客户端连接到所有可用的后端。我应该WeightedLoadbalanceStrategy改用吗？或者discoveryClient抽象应该每次只返回一个服务器——但这不再是一个“池”客户端，对吧？

也许我应该从总体上重新考虑方法。我有几十万个客户端，所以我想平衡后端的负载 - 将其分布在后端的多个实例中，因此每个客户端随机连接到后端的一个实例，但能够重新连接到另一个实例，如果实例连接失败。我认为将所有客户端同时连接到每个后端实例并不是一个好主意，但也许我错了？

于 2020-12-14T16:56:26.277 回答