聊聊WebClient的LoadBalance支持

为什么80%的码农都做不了架构师？>>>   

序

本文主要研究一下WebClient的LoadBalance支持

代码实例

配置

@Configuration public class WebClientConfig {@Autowiredprivate LoadBalancerExchangeFilterFunction lbFunction;@Beanpublic WebClient webClient(){return WebClient.builder().filter(lbFunction).build();} }

请求department-service

@Component public class DepartmentService {@AutowiredWebClient webClient;public Flux<Department> getDepartmentsByOrgId(Long orgId) {return webClient.get().uri("http://department-service/organization/{orgId}",orgId).retrieve().bodyToFlux(Department.class);} }

controller

@AutowiredDepartmentService departmentService;@GetMapping("/departments")public Flux<Department> getDepartmentsById(Long orgId){return departmentService.getDepartmentsByOrgId(orgId);}

/flux/departments?orgId=1

[{"id": 1,"name": "department 1","employees": []} ]

异常情况

Connection refused

2018-04-29 13:09:15 ERROR [organization-service,,,] Failed to handle request [GET http://localhost:8092/flux/departments?orgId=1] io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: /172.16.205.106:8091at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[na:1.8.0_151]at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717) ~[na:1.8.0_151]at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:325) ~[netty-transport-4.1.23.Final.jar:4.1.23.Final]at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:340) ~[netty-transport-4.1.23.Final.jar:4.1.23.Final]at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:633) ~[netty-transport-4.1.23.Final.jar:4.1.23.Final]at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580) ~[netty-transport-4.1.23.Final.jar:4.1.23.Final]at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497) ~[netty-transport-4.1.23.Final.jar:4.1.23.Final]at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459) ~[netty-transport-4.1.23.Final.jar:4.1.23.Final]at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886) ~[netty-common-4.1.23.Final.jar:4.1.23.Final]at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_151] Caused by: java.net.ConnectException: Connection refused... 10 common frames omitted

instance can not be null

2018-04-29 13:12:08 ERROR [organization-service,,,] Failed to handle request [GET http://localhost:8092/flux/departments?orgId=1] java.lang.IllegalArgumentException: instance can not be nullat org.springframework.util.Assert.notNull(Assert.java:193) ~[spring-core-5.0.5.RELEASE.jar:5.0.5.RELEASE]at org.springframework.cloud.netflix.ribbon.RibbonLoadBalancerClient.reconstructURI(RibbonLoadBalancerClient.java:53) ~[spring-cloud-netflix-ribbon-2.0.0.RC1.jar:2.0.0.RC1]at org.springframework.cloud.client.loadbalancer.reactive.LoadBalancerExchangeFilterFunction.filter(LoadBalancerExchangeFilterFunction.java:34) ~[spring-cloud-commons-2.0.0.RC1.jar:2.0.0.RC1]at org.springframework.web.reactive.function.client.ExchangeFilterFunction.lambda$andThen$1(ExchangeFilterFunction.java:56) ~[spring-webflux-5.0.5.RELEASE.jar:5.0.5.RELEASE]at org.springframework.web.reactive.function.client.ExchangeFilterFunction.lambda$apply$2(ExchangeFilterFunction.java:67) ~[spring-webflux-5.0.5.RELEASE.jar:5.0.5.RELEASE]at org.springframework.web.reactive.function.client.DefaultWebClient$DefaultRequestBodyUriSpec.exchange(DefaultWebClient.java:320) ~[spring-webflux-5.0.5.RELEASE.jar:5.0.5.RELEASE]at org.springframework.web.reactive.function.client.DefaultWebClient$DefaultRequestBodyUriSpec.retrieve(DefaultWebClient.java:367) ~[spring-webflux-5.0.5.RELEASE.jar:5.0.5.RELEASE]at com.example.service.DepartmentService.getDepartmentsByOrgId(DepartmentService.java:24) ~[classes/:na]at com.example.controller.FluxDemoController.getDepartmentsById(FluxDemoController.java:23) ~[classes/:na]at sun.reflect.GeneratedMethodAccessor82.invoke(Unknown Source) ~[na:na]at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_151]at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_151]at org.springframework.web.reactive.result.method.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:243) ~[spring-webflux-5.0.5.RELEASE.jar:5.0.5.RELEASE]at org.springframework.web.reactive.result.method.InvocableHandlerMethod.lambda$invoke$0(InvocableHandlerMethod.java:138) ~[spring-webflux-5.0.5.RELEASE.jar:5.0.5.RELEASE]at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:118) [reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1073) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoZip$ZipCoordinator.signal(MonoZip.java:241) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoZip$ZipInner.onNext(MonoZip.java:323) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:81) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:185) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:81) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.FluxDefaultIfEmpty$DefaultIfEmptySubscriber.onNext(FluxDefaultIfEmpty.java:92) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:81) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:81) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1630) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.request(ScopePassingSpanSubscriber.java:69) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1444) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:1318) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onSubscribe(ScopePassingSpanSubscriber.java:63) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoDefaultIfEmpty.subscribe(MonoDefaultIfEmpty.java:37) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoPeek.subscribe(MonoPeek.java:71) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Mono.subscribe(Mono.java:3080) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:148) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoPeek.subscribe(MonoPeek.java:71) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150) [reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:81) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:81) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:76) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:81) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:271) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:803) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:81) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1630) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.request(ScopePassingSpanSubscriber.java:69) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1444) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:1318) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onSubscribe(ScopePassingSpanSubscriber.java:63) [spring-cloud-sleuth-core-2.0.0.RC1.jar:2.0.0.RC1]at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Mono.subscribe(Mono.java:3080) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:418) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onSubscribe(FluxConcatMap.java:210) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:128) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:61) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.FluxConcatMap.subscribe(FluxConcatMap.java:121) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.FluxLift.subscribe(FluxLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoNext.subscribe(MonoNext.java:40) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150) [reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1073) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoZip$ZipCoordinator.signal(MonoZip.java:241) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoZip$ZipInner.onNext(MonoZip.java:323) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1630) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoZip$ZipInner.onSubscribe(MonoZip.java:312) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Mono.subscribe(Mono.java:3080) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Mono.subscribe(Mono.java:3080) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:167) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoSubscriberContext.subscribe(MonoSubscriberContext.java:47) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoPeek.subscribe(MonoPeek.java:71) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.Mono.subscribe(Mono.java:3080) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:167) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:70) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.core.publisher.MonoLift.subscribe(MonoLift.java:46) ~[reactor-core-3.1.6.RELEASE.jar:3.1.6.RELEASE]at reactor.ipc.netty.channel.ChannelOperations.applyHandler(ChannelOperations.java:381) ~[reactor-netty-0.7.6.RELEASE.jar:0.7.6.RELEASE]at reactor.ipc.netty.http.server.HttpServerOperations.onHandlerStart(HttpServerOperations.java:397) ~[reactor-netty-0.7.6.RELEASE.jar:0.7.6.RELEASE]at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163) ~[netty-common-4.1.23.Final.jar:4.1.23.Final]at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:404) ~[netty-common-4.1.23.Final.jar:4.1.23.Final]at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:463) ~[netty-transport-4.1.23.Final.jar:4.1.23.Final]at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886) ~[netty-common-4.1.23.Final.jar:4.1.23.Final]at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_151]

源码解析

LoadBalancerExchangeFilterFunction

spring-cloud-commons-2.0.0.RC1-sources.jar!/org/springframework/cloud/client/loadbalancer/reactive/LoadBalancerExchangeFilterFunction.java

public class LoadBalancerExchangeFilterFunction implements ExchangeFilterFunction {private final LoadBalancerClient loadBalancerClient;public LoadBalancerExchangeFilterFunction(LoadBalancerClient loadBalancerClient) {this.loadBalancerClient = loadBalancerClient;}@Overridepublic Mono<ClientResponse> filter(ClientRequest request, ExchangeFunction next) {URI originalUrl = request.url();String serviceId = originalUrl.getHost();Assert.state(serviceId != null, "Request URI does not contain a valid hostname: " + originalUrl);//TODO: reactive lb clientServiceInstance instance = this.loadBalancerClient.choose(serviceId);URI uri = this.loadBalancerClient.reconstructURI(instance, originalUrl);ClientRequest newRequest = ClientRequest.method(request.method(), uri).headers(headers -> headers.addAll(request.headers())).cookies(cookies -> cookies.addAll(request.cookies())).attributes(attributes -> attributes.putAll(request.attributes())).body(request.body()).build();return next.exchange(newRequest);}}

对于webclient来说，在这个filterChain中使用了LoadBalancerExchangeFilterFunction，可以看到使用了LoadBalancerExchangeFilterFunction的filter方法里头，对原来的request进行了包装，使用loadBalancerClient根据服务ID进行服务发现选取可用的服务地址，然后替换原来的uri，构造成新的request传递到下一个filter

DefaultWebClientBuilder

spring-webflux-5.0.5.RELEASE-sources.jar!/org/springframework/web/reactive/function/client/DefaultWebClientBuilder.java

class DefaultWebClientBuilder implements WebClient.Builder {@Nullableprivate List<ExchangeFilterFunction> filters;@Overridepublic WebClient.Builder filter(ExchangeFilterFunction filter) {Assert.notNull(filter, "ExchangeFilterFunction must not be null");initFilters().add(filter);return this;}@Overridepublic WebClient build() {ExchangeFunction exchange = initExchangeFunction();ExchangeFunction filteredExchange = (this.filters != null ? this.filters.stream().reduce(ExchangeFilterFunction::andThen).map(filter -> filter.apply(exchange)).orElse(exchange) : exchange);return new DefaultWebClient(filteredExchange, initUriBuilderFactory(),unmodifiableCopy(this.defaultHeaders), unmodifiableCopy(this.defaultCookies),new DefaultWebClientBuilder(this));}

可以看到调用webClient的filter就会往filters添加，之后在build的时候，利用ExchangeFilterFunction::andThen构造一个ExchangeFunction，传递给DefaultWebClient的构造器

ExchangeFilterFunction

spring-webflux-5.0.5.RELEASE-sources.jar!/org/springframework/web/reactive/function/client/ExchangeFilterFunction.java

/*** Apply this filter to the given request and exchange function.* <p>The given {@linkplain ExchangeFunction exchange function} represents the next entity* in the chain, and can be {@linkplain ExchangeFunction#exchange(ClientRequest) invoked}* in order to proceed to the exchange, or not invoked to block the chain.* @param request the request* @param next the next exchange function in the chain* @return the filtered response*/Mono<ClientResponse> filter(ClientRequest request, ExchangeFunction next);/*** Return a composed filter function that first applies this filter, and then applies the* {@code after} filter.* @param after the filter to apply after this filter is applied* @return a composed filter that first applies this function and then applies the* {@code after} function*/default ExchangeFilterFunction andThen(ExchangeFilterFunction after) {Assert.notNull(after, "'after' must not be null");return (request, next) -> {ExchangeFunction nextExchange = exchangeRequest -> after.filter(exchangeRequest, next);return filter(request, nextExchange);};}/*** Apply this filter to the given exchange function, resulting in a filtered exchange function.* @param exchange the exchange function to filter* @return the filtered exchange function*/default ExchangeFunction apply(ExchangeFunction exchange) {Assert.notNull(exchange, "'exchange' must not be null");return request -> this.filter(request, exchange);}

ExchangeFilterFunction通过andThen添加filter链(里头的lambda是实现filter这个函数式方法)，最后通过apply转换为ExchangeFunction

小结

webClient的loadBalanced支持比restTemplate更为简洁和清晰，直接使用filter模式，通过loadBalancerClient获取服务地址，替换uri再传递给下一个filter。如果loadBalancerClient没能获得到服务地址的话，则RibbonLoadBalancerClient.reconstructURI方法会报错java.lang.IllegalArgumentException: instance can not be null。另外由于注册中心的信息可能有延迟，因为也可能存在Connection refused的异常。

doc

• Spring WebFlux WebClient as a Load Balancer Client

转载于:https://my.oschina.net/go4it/blog/1803113

总结

以上是生活随笔为你收集整理的聊聊WebClient的LoadBalance支持的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。