使用 Spring 的 RSocketRequester 正确使用 LoadbalanceRSocketClient
Correct usage of LoadbalanceRSocketClient with Spring's RSocketRequester
我试图在 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[=13=](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]
我怀疑我没有正确配置请求程序或没有正确使用它。感谢任何提示,因为文档和测试在这方面似乎很薄。
理想情况下,我希望客户端在 server/connectivity 失败时透明地切换到任何下一个可用服务器。现在重新连接尝试似乎只发生在下一次调用 timer()
方法时,这并不理想,因为客户端需要处理来自服务器的传入消息。我观察到的另一件事是,即使如此 "/foo"
也是 FnF 路由,除非我在 send()
服务器从未收到呼叫后执行 block()
。
关于 FnF 的问题,这是 Rx 模型的一部分。没有订阅,事件就不会发生。您可以在订阅前自由调用 API 返回没有副作用的 Mono,任何其他行为都是错误。
/**
* 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 次。
不断更新端点列表
LoadbalanceClient
旨在与负责保持 List
活动 Instance
的发现服务集成。也就是说,如果其中一项服务从集群中消失,Discovery 服务会更新其 List
可用 Instance
s。
另一方面,要实现客户端负载均衡,我们必须知道集群中可用服务的列表。很明显,要设置负载均衡,我们可以检索服务列表并将其提供给负载均衡器 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());
然而,假设我们处于一个完全分布式的环境中,现在每一个服务消失并再次出现 - 运行在绝对新的主机和端口上(例如,不依赖特定 IP 地址的 kubernates 集群)。也就是说,Loadbalancing 必须考虑这种情况,为了避免池中出现死节点,它会从池中完全删除不健康的节点。
现在,如果所有节点消失并在一段时间后出现,它们将不再包含在池中(如果提供更新的 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> ...
并在不同的地方自动装配它,并且 subscribe
多次无需担心每个 Mono<RSocket>.subscribe
的结果 RSocket instance
都会不同。另一方面,.reconnect
,如果给定的 RSocket
断开连接(例如失去连接的情况),那么下一次订阅这样的 Mono<RSocket>
将只能抵抗一次新的 RSocket
并发 .subscribe
个调用。
虽然这听起来很有用的功能,但在 RSocketPool
中我们并不太依赖它并且只使用 Mono<RSocket>
一次来解析和缓存 RSocketPool 中的 RSocket 实例。也就是说,如果这样的 RSocket 将断开连接,我们将不会尝试再次订阅给定的 Mono<RSocket>
(我们假设设置的主机和端口将被更改)
Oleh,我尝试了你的建议并且在一定程度上有效,虽然我仍然不能完全得到我需要的行为。
我想做的是:
- 客户端一次连接到单个(随机)后端
- 如果后端或与后端的连接失败,客户端应尝试连接到下一个可用的后端。
我想我不能使用 RoundRobinLoadbalanceStrategy
,因为它将客户端连接到所有可用的后端。我应该改用 WeightedLoadbalanceStrategy
吗?或者应该 discoveryClient
每次只抽象 return 一个服务器 - 但那不再是 'pool' 客户端,对吧?
也许我应该从总体上重新考虑方法。我有几十万个客户端,所以我想平衡后端的负载 - 将其分布在后端的多个实例中,因此每个客户端随机连接到后端的一个实例,但能够重新连接到另一个实例实例,如果实例连接失败。我认为同时将所有客户端连接到每个后端实例不是一个好主意,但也许我错了?
我试图在 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[=13=](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]
我怀疑我没有正确配置请求程序或没有正确使用它。感谢任何提示,因为文档和测试在这方面似乎很薄。
理想情况下,我希望客户端在 server/connectivity 失败时透明地切换到任何下一个可用服务器。现在重新连接尝试似乎只发生在下一次调用 timer()
方法时,这并不理想,因为客户端需要处理来自服务器的传入消息。我观察到的另一件事是,即使如此 "/foo"
也是 FnF 路由,除非我在 send()
服务器从未收到呼叫后执行 block()
。
关于 FnF 的问题,这是 Rx 模型的一部分。没有订阅,事件就不会发生。您可以在订阅前自由调用 API 返回没有副作用的 Mono,任何其他行为都是错误。
/**
* 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 次。
不断更新端点列表
LoadbalanceClient
旨在与负责保持 List
活动 Instance
的发现服务集成。也就是说,如果其中一项服务从集群中消失,Discovery 服务会更新其 List
可用 Instance
s。
另一方面,要实现客户端负载均衡,我们必须知道集群中可用服务的列表。很明显,要设置负载均衡,我们可以检索服务列表并将其提供给负载均衡器 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());
然而,假设我们处于一个完全分布式的环境中,现在每一个服务消失并再次出现 - 运行在绝对新的主机和端口上(例如,不依赖特定 IP 地址的 kubernates 集群)。也就是说,Loadbalancing 必须考虑这种情况,为了避免池中出现死节点,它会从池中完全删除不健康的节点。
现在,如果所有节点消失并在一段时间后出现,它们将不再包含在池中(如果提供更新的 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> ...
并在不同的地方自动装配它,并且 subscribe
多次无需担心每个 Mono<RSocket>.subscribe
的结果 RSocket instance
都会不同。另一方面,.reconnect
,如果给定的 RSocket
断开连接(例如失去连接的情况),那么下一次订阅这样的 Mono<RSocket>
将只能抵抗一次新的 RSocket
并发 .subscribe
个调用。
虽然这听起来很有用的功能,但在 RSocketPool
中我们并不太依赖它并且只使用 Mono<RSocket>
一次来解析和缓存 RSocketPool 中的 RSocket 实例。也就是说,如果这样的 RSocket 将断开连接,我们将不会尝试再次订阅给定的 Mono<RSocket>
(我们假设设置的主机和端口将被更改)
Oleh,我尝试了你的建议并且在一定程度上有效,虽然我仍然不能完全得到我需要的行为。
我想做的是:
- 客户端一次连接到单个(随机)后端
- 如果后端或与后端的连接失败,客户端应尝试连接到下一个可用的后端。
我想我不能使用 RoundRobinLoadbalanceStrategy
,因为它将客户端连接到所有可用的后端。我应该改用 WeightedLoadbalanceStrategy
吗?或者应该 discoveryClient
每次只抽象 return 一个服务器 - 但那不再是 'pool' 客户端,对吧?
也许我应该从总体上重新考虑方法。我有几十万个客户端,所以我想平衡后端的负载 - 将其分布在后端的多个实例中,因此每个客户端随机连接到后端的一个实例,但能够重新连接到另一个实例实例,如果实例连接失败。我认为同时将所有客户端连接到每个后端实例不是一个好主意,但也许我错了?