共计 9426 个字符,预计需要花费 24 分钟才能阅读完成。
本系列代码地址:https://github.com/JoJoTec/sp…
要想实现咱们上一节中提到的:
- 须要在重试以及断路中加一些日志,便于日后的优化
- 须要定义重试的 Exception,并且与断路器相结合,将非 2xx 的响应码也封装成特定的异样
- 须要在断路器相干的 Operator 中减少相似于 FeignClient 中的负载平衡的数据更新,使得负载平衡更加智能
咱们须要将 resilience4j 自身提供的粘合库做一些革新,其实次要就是对 resilience4j 实现的 project reactor 的 Operator 进行革新。
对于断路器的革新
首先,WebClient 的返回对象只可能是 ClientResponse
类型,所以咱们这里革新进去的 Operator 不用带上形参,只须要针对 ClientResponse 即可,即:
public class ClientResponseCircuitBreakerOperator implements UnaryOperator<Publisher<ClientResponse>> {...}
在原有的断路器逻辑中,咱们须要退出针对 GET 办法以及之前定义的能够重试的门路匹配配置能够重试的逻辑,这须要咱们拿到原有申请的 URL 信息。然而 ClientResponse 中并没有裸露这些信息的接口,其默认实现 DefaultClientResponse(咱们只有没有本人给 WebClient 退出非凡的革新逻辑,实现都是 DefaultClientResponse)中的 request()
办法能够获取申请 HttpRequest,其中蕴含 url 信息。然而这个类还有办法都是 package-private 的,咱们须要反射进去:
ClientResponseCircuitBreakerSubscriber
private static final Class<?> aClass;
private static final Method request;
static {
try {aClass = Class.forName("org.springframework.web.reactive.function.client.DefaultClientResponse");
request = ReflectionUtils.findMethod(aClass, "request");
request.setAccessible(true);
} catch (Exception e) {throw new RuntimeException(e);
}
}
之后,在获取到 ClientResponse 之后记录断路器的逻辑中,须要退出下面提到的对于重试的革新,以及负载均衡器的记录:
ClientResponseCircuitBreakerSubscriber
protected void hookOnNext(ClientResponse clientResponse) {if (!isDisposed()) {if (singleProducer && successSignaled.compareAndSet(false, true)) {int rawStatusCode = clientResponse.rawStatusCode();
HttpStatus httpStatus = HttpStatus.resolve(rawStatusCode);
try {HttpRequest httpRequest = (HttpRequest) request.invoke(clientResponse);
// 判断办法是否为 GET,以及是否在可重试门路配置中,从而得出是否能够重试
if (httpRequest.getMethod() != HttpMethod.GET && !webClientProperties.retryablePathsMatch(httpRequest.getURI().getPath())) {
// 如果不能重试,则间接返回后果
circuitBreaker.onResult(circuitBreaker.getCurrentTimestamp() - start, circuitBreaker.getTimestampUnit(), clientResponse);
} else {if (httpStatus != null && httpStatus.is2xxSuccessful()) {
// 如果胜利,则间接返回后果
circuitBreaker.onResult(circuitBreaker.getCurrentTimestamp() - start, circuitBreaker.getTimestampUnit(), clientResponse);
} else {
/**
* 如果异样,参考 DefaultClientResponse 的代码进行异样封装
* @see org.springframework.web.reactive.function.client.DefaultClientResponse#createException
*/
Exception exception;
if (httpStatus != null) {exception = WebClientResponseException.create(rawStatusCode, httpStatus.getReasonPhrase(), clientResponse.headers().asHttpHeaders(), EMPTY, null, null);
} else {exception = new UnknownHttpStatusCodeException(rawStatusCode, clientResponse.headers().asHttpHeaders(), EMPTY, null, null);
}
circuitBreaker.onError(circuitBreaker.getCurrentTimestamp() - start, circuitBreaker.getTimestampUnit(), exception);
downstreamSubscriber.onError(exception);
return;
}
}
} catch (Exception e) {log.fatal("judge request method in circuit breaker error! the resilience4j feature would not be enabled: {}", e.getMessage(), e);
circuitBreaker.onResult(circuitBreaker.getCurrentTimestamp() - start, circuitBreaker.getTimestampUnit(), clientResponse);
}
}
eventWasEmitted.set(true);
downstreamSubscriber.onNext(clientResponse);
}
}
同样的,在原有的实现,勾销还有失败的记录逻辑中,也加上记录负载平衡数据:
ClientResponseCircuitBreakerSubscriber
@Override
protected void hookOnComplete() {if (successSignaled.compareAndSet(false, true)) {serviceInstanceMetrics.recordServiceInstanceCalled(serviceInstance, true);
circuitBreaker.onSuccess(circuitBreaker.getCurrentTimestamp() - start, circuitBreaker.getTimestampUnit());
}
downstreamSubscriber.onComplete();}
@Override
public void hookOnCancel() {if (!successSignaled.get()) {serviceInstanceMetrics.recordServiceInstanceCalled(serviceInstance, true);
if (eventWasEmitted.get()) {circuitBreaker.onSuccess(circuitBreaker.getCurrentTimestamp() - start, circuitBreaker.getTimestampUnit());
} else {circuitBreaker.releasePermission();
}
}
}
@Override
protected void hookOnError(Throwable e) {serviceInstanceMetrics.recordServiceInstanceCalled(serviceInstance, false);
circuitBreaker.onError(circuitBreaker.getCurrentTimestamp() - start, circuitBreaker.getTimestampUnit(), e);
downstreamSubscriber.onError(e);
}
粘合 WebClient 与 resilience4j 的同时笼罩重试逻辑
因为后面的断路器中,咱们针对能够重试的非 2XX 响应封装成为 WebClientResponseException。所以在重试器中,咱们须要加上针对这个异样的重试。
同时,须要将 重试器 放在 负载均衡器 之前,因为每次重试,都要从负载均衡器中获取一个新的实例。同时,断路器 须要放在 负载均衡器 之后,因为只有在这个之后,能力获取到本次调用的实例,咱们的的断路器是针对实例办法级别的:
WebClientDefaultConfiguration.java
@Bean
public WebClient getWebClient(
ReactorLoadBalancerExchangeFilterFunction lbFunction,
WebClientConfigurationProperties webClientConfigurationProperties,
Environment environment,
RetryRegistry retryRegistry,
CircuitBreakerRegistry circuitBreakerRegistry,
ServiceInstanceMetrics serviceInstanceMetrics
) {String name = environment.getProperty(WebClientNamedContextFactory.PROPERTY_NAME);
Map<String, WebClientConfigurationProperties.WebClientProperties> configs = webClientConfigurationProperties.getConfigs();
if (configs == null || configs.size() == 0) {throw new BeanCreationException("Failed to create webClient, please provide configurations under namespace: webclient.configs");
}
WebClientConfigurationProperties.WebClientProperties webClientProperties = configs.get(name);
if (webClientProperties == null) {throw new BeanCreationException("Failed to create webClient, please provide configurations under namespace: webclient.configs." + name);
}
String serviceName = webClientProperties.getServiceName();
// 如果没填写微服务名称,就应用配置 key 作为微服务名称
if (StringUtils.isBlank(serviceName)) {serviceName = name;}
String baseUrl = webClientProperties.getBaseUrl();
// 如果没填写 baseUrl,就应用微服务名称填充
if (StringUtils.isBlank(baseUrl)) {baseUrl = "http://" + serviceName;}
Retry retry = null;
try {retry = retryRegistry.retry(serviceName, serviceName);
} catch (ConfigurationNotFoundException e) {retry = retryRegistry.retry(serviceName);
}
// 笼罩其中的异样判断
retry = Retry.of(serviceName, RetryConfig.from(retry.getRetryConfig()).retryOnException(throwable -> {
//WebClientResponseException 会重试,因为在这里能 catch 的 WebClientResponseException 只对能够重试的申请封装了 WebClientResponseException
// 参考 ClientResponseCircuitBreakerSubscriber 的代码
if (throwable instanceof WebClientResponseException) {log.info("should retry on {}", throwable.toString());
return true;
}
// 断路器异样重试,因为申请没有收回去
if (throwable instanceof CallNotPermittedException) {log.info("should retry on {}", throwable.toString());
return true;
}
if (throwable instanceof WebClientRequestException) {WebClientRequestException webClientRequestException = (WebClientRequestException) throwable;
HttpMethod method = webClientRequestException.getMethod();
URI uri = webClientRequestException.getUri();
// 判断是否为响应超时,响应超时代表申请曾经收回去了,对于非 GET 并且没有标注能够重试的申请则不能重试
boolean isResponseTimeout = false;
Throwable cause = throwable.getCause();
//netty 的读取超时个别是 ReadTimeoutException
if (cause instanceof ReadTimeoutException) {log.info("Cause is a ReadTimeoutException which indicates it is a response time out");
isResponseTimeout = true;
} else {
// 对于其余一些框架,应用了 java 底层 nio 的个别是 SocketTimeoutException,message 为 read time out
// 还有一些其余异样,然而 message 都会有 read time out 字段,所以通过 message 判断
String message = throwable.getMessage();
if (StringUtils.isNotBlank(message) && StringUtils.containsIgnoreCase(message.replace("",""), "readtimeout")) {log.info("Throwable message contains readtimeout which indicates it is a response time out");
isResponseTimeout = true;
}
}
// 如果申请是 GET 或者标注了重试,则直接判断能够重试
if (method == HttpMethod.GET || webClientProperties.retryablePathsMatch(uri.getPath())) {log.info("should retry on {}-{}, {}", method, uri, throwable.toString());
return true;
} else {
// 否则,只针对申请还没有收回去的异样进行重试
if (isResponseTimeout) {log.info("should not retry on {}-{}, {}", method, uri, throwable.toString());
} else {log.info("should retry on {}-{}, {}", method, uri, throwable.toString());
return true;
}
}
}
return false;
}).build());
HttpClient httpClient = HttpClient
.create()
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, (int) webClientProperties.getConnectTimeout().toMillis())
.doOnConnected(connection ->
connection
.addHandlerLast(new ReadTimeoutHandler((int) webClientProperties.getResponseTimeout().toSeconds()))
.addHandlerLast(new WriteTimeoutHandler((int) webClientProperties.getResponseTimeout().toSeconds()))
);
Retry finalRetry = retry;
String finalServiceName = serviceName;
return WebClient.builder()
.exchangeStrategies(ExchangeStrategies.builder()
.codecs(configurer -> configurer
.defaultCodecs()
// 最大 body 占用 16m 内存
.maxInMemorySize(16 * 1024 * 1024))
.build())
.clientConnector(new ReactorClientHttpConnector(httpClient))
//Retry 在负载平衡前
.filter((clientRequest, exchangeFunction) -> {
return exchangeFunction
.exchange(clientRequest)
.transform(ClientResponseRetryOperator.of(finalRetry));
})
// 负载均衡器,改写 url
.filter(lbFunction)
// 实例级别的断路器须要在负载平衡获取真正地址之后
.filter((clientRequest, exchangeFunction) -> {ServiceInstance serviceInstance = getServiceInstance(clientRequest);
serviceInstanceMetrics.recordServiceInstanceCall(serviceInstance);
CircuitBreaker circuitBreaker;
// 这时候的 url 是通过负载均衡器的,是实例的 url
// 须要留神的一点是,应用异步 client 的时候,最好不要带门路参数,否则这里的断路器成果不好
// 断路器是每个实例每个门路一个断路器
String instancId = clientRequest.url().getHost() + ":" + clientRequest.url().getPort() + clientRequest.url().getPath();
try {
// 应用实例 id 新建或者获取现有的 CircuitBreaker, 应用 serviceName 获取配置
circuitBreaker = circuitBreakerRegistry.circuitBreaker(instancId, finalServiceName);
} catch (ConfigurationNotFoundException e) {circuitBreaker = circuitBreakerRegistry.circuitBreaker(instancId);
}
log.info("webclient circuit breaker [{}-{}] status: {}, data: {}", finalServiceName, instancId, circuitBreaker.getState(), JSON.toJSONString(circuitBreaker.getMetrics()));
return exchangeFunction.exchange(clientRequest).transform(ClientResponseCircuitBreakerOperator.of(circuitBreaker, serviceInstance, serviceInstanceMetrics, webClientProperties));
}).baseUrl(baseUrl)
.build();}
private ServiceInstance getServiceInstance(ClientRequest clientRequest) {URI url = clientRequest.url();
DefaultServiceInstance defaultServiceInstance = new DefaultServiceInstance();
defaultServiceInstance.setHost(url.getHost());
defaultServiceInstance.setPort(url.getPort());
return defaultServiceInstance;
}
这样,咱们就实现了咱们封装的基于配置的 WebClient
微信搜寻“我的编程喵”关注公众号,每日一刷,轻松晋升技术,斩获各种 offer: