Spring5 WebClient的超时负载均衡异常重试

Posted 2021-04-15 CodingWithFun

WebClient简介

WebClient是从Spring WebFlux 5.0版本开始提供的一个非阻塞的基于响应式编程的进行Http请求的客户端工具。其基本使用比较简单，本文介绍其一些高级特性。

超时设置

public static void main(String[] args) throws IOException { WebClient webClient = WebClient.create(); Mono<String> response = webClient.get() .uri("http://192.168.101.100:8080")  .retrieve() .bodyToMono(String.class).timeout(Duration.ofSeconds(3)); response.subscribe();
 /**为了让当前进程不退出*/ System.in.read();}

运行输出如下日志：

从中我们可以看到，3秒之后出现超时异常，但这是Netty层仍在尝试连接，之后经过30秒出现io.netty.channel.ConnectTimeoutException异常。所以使用timeout方法仅仅是在Reactor层设置时间超时，底层的网络连接还没有关闭，默认需经过30秒才能正常结束。

故推荐使用以下方式在Netty层直接设置超时，这样可以及时释放连接：

ReactorClientHttpConnector connector = new ReactorClientHttpConnector( options -> options.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000) .afterNettyContextInit(ctx -> { ctx.addHandlerLast(new ReadTimeoutHandler(2000, TimeUnit.MILLISECONDS)); ctx.addHandlerLast(new WriteTimeoutHandler(2000, TimeUnit.MILLISECONDS)); }));
WebClient webClient = WebClient.builder().clientConnector(connector).build();

负载均衡

Spring Cloud提供了LoadBalancerExchangeFilterFunction，以过滤器的方式实现负载均衡。其原理是从注册中心获取服务实例，以轮询算法选取实例，将实例的IP与端口替换请求url中的服务名。

本质是就是一种客户端负载均衡机制。其使用一般比较简单，直接在配置类中注入，生成WebClient或WebClient.Builder的Bean，然后在需要的地方注入该Bean即可：

@Beanpublic WebClient.Builder builder(LoadBalancerExchangeFilterFunction lbFunction) { return WebClient.builder().filter(lbFunction).clientConnector(connector);}

异常重试

WebClient返回Mono或Flux对象，服务调用时可能由于网络抖动暂时失败。这时我们可以使用retry或retryBackoff方法进行失败重试，实例代码如下：

/*重试两次，中间没有间隔*/Mono<String> response = client.get() .uri("http://SERVICE-A") .retrieve().bodyToMono(String.class).retry(2);
/*重试两次，间隔100ms*/Mono<String> backoff = client.get() .uri("http://SERVICE-A") .retrieve().bodyToMono(String.class) .retryBackoff(2, Duration.ofMillis(100));

但有时可能真的是服务实例宕机或者整个机器都宕机了，这时再用retry方法就不好使了，即使这个服务还有其他可用实例。因为重试时WebClient的url并不会改变，请求不能转发给其他实例。这种情况，尤其是在使用Eureka注册中心时很常见，因为服务宕机Eureka只能通过超时机制将其剔除。

这时我们可以使用Mono提供的其他错误接口进行失败重试。还是分析上述场景，对于机器宕机，这时TCP连接会抛出ConnectTimeoutException。对于实例宕机，但机器正常情况下，这时会抛出java.net.ConnectException异常。而且这两个异常有共同的父类java.net.SocketException。故我们只需要用Mono提供的onErrorResume函数处理SocketException异常，并进行重定向到其他实例即可：

Mono<String> backoff = client.get() .uri("http://192.168.98.111") .retrieve().bodyToMono(String.class).onErrorResume(SocketException.class, re -> client.get().uri("http://192.168.98.112").retrieve().bodyToMono(String.class) ).onErrorReturn("{code: -1, msg: '服务异常'}");

以上代码简答解释一下：首先向192.168.98.111实例发送请求，抛出SocketException异常，接着向192.168.98.112发出请求，如果正常返回结果，如果不幸的是仍然异常，那么返回一个默认结果。

如果结合注册中心，并使用负载均衡机制就更简单了，不需要手动填入IP,

Mono<String> backoff = client.get() .uri("http://SERVCIE-A") .retrieve().bodyToMono(String.class).onErrorResume(SocketException.class, re -> client.get().uri("http://SERVCIE-A").retrieve().bodyToMono(String.class) ) .onErrorReturn("{code: -1, msg: '服务异常'}");

整个流程如下如所示：