gpt4 book ai didi

spring-messaging - 使用 Spring 的 RSocketRequester 正确使用 LoadbalanceRSocketClient

转载 作者:行者123 更新时间:2023-12-04 15:57:54 26 4
gpt4 key购买 nike

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

我有两个 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 路由,除非我在 send() 服务器之后执行 block()从不接听电话。

最佳答案

不断更新端点列表

LoadbalanceClient旨在与负责保存 List 的 Discovery 服务集成。活着的Instance s。也就是说,如果其中一项服务从集群中消失,Discovery 服务会更新其 List可用 Instance秒。

另一方面,要实现客户端负载平衡,我们必须知道集群中可用服务的列表。很明显,要设置负载平衡,我们可以检索服务列表并将其提供给 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>>尚未完成,最终可能会提供新的更新。

Note, the implementation is smart enough to keep active nodes on every update from the discovery service. That said if there is such a service instance in the pool, you will not get 2 connections at the same time.

关于重新连接功能的附注

您可能会注意到,RSocketConnector提供了一个很棒的功能,叫做 .reconnect .乍一看,reconnect 的用法似乎将使您的连接保持畅通无阻。不幸的是,事实并非如此。 .reconnect功能旨在让您的 Mono<RSocket>可重用缓存语义,这意味着您可以创建 @Bean Mono<RSocket> ...并在不同的地方 Autowiring 它和subscribe多次而不用担心结果RSocket instance每个 Mono<RSocket>.subscribe 都会有所不同.另一方面,.reconnect , 如果给定 RSocket下一次订阅此类 Mono<RSocket> 时会断开连接(例如失去连接的情况)将抵抗一个新的RSocket所有并发 .subscribe 仅一次来电。

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

关于spring-messaging - 使用 Spring 的 RSocketRequester 正确使用 LoadbalanceRSocketClient,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/65254112/

26 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com