聊聊spring cloud的AbstractLoadBalancingClient

序

本文主要研究一下spring cloud的AbstractLoadBalancingClient

AbstractLoadBalancingClient

spring-cloud-netflix-ribbon-2.0.0.RELEASE-sources.jar!/org/springframework/cloud/netflix/ribbon/support/AbstractLoadBalancingClient.java

/**

* @author Spencer Gibb

*/

public abstract class AbstractLoadBalancingClient extends

AbstractLoadBalancerAwareClient implements ServiceInstanceChooser {

protected int connectTimeout;

protected int readTimeout;

protected boolean secure;

protected boolean followRedirects;

protected boolean okToRetryOnAllOperations;

protected final D delegate;

protected final IClientConfig config;

protected final ServerIntrospector serverIntrospector;

public boolean isClientRetryable(ContextAwareRequest request) {

return false;

}

protected AbstractLoadBalancingClient(IClientConfig config, ServerIntrospector serverIntrospector) {

super(null);

this.delegate = createDelegate(config);

this.config = config;

this.serverIntrospector = serverIntrospector;

this.setRetryHandler(RetryHandler.DEFAULT);

initWithNiwsConfig(config);

}

protected AbstractLoadBalancingClient(D delegate, IClientConfig config, ServerIntrospector serverIntrospector) {

super(null);

this.delegate = delegate;

this.config = config;

this.serverIntrospector = serverIntrospector;

this.setRetryHandler(RetryHandler.DEFAULT);

initWithNiwsConfig(config);

}

@Override

public void initWithNiwsConfig(IClientConfig clientConfig) {

super.initWithNiwsConfig(clientConfig);

RibbonProperties ribbon = RibbonProperties.from(clientConfig);

this.connectTimeout = ribbon.connectTimeout(DEFAULT_CONNECT_TIMEOUT);

this.readTimeout = ribbon.readTimeout(DEFAULT_READ_TIMEOUT);

this.secure = ribbon.isSecure();

this.followRedirects = ribbon.isFollowRedirects();

this.okToRetryOnAllOperations = ribbon.isOkToRetryOnAllOperations();

}

protected abstract D createDelegate(IClientConfig config);

public D getDelegate() {

return this.delegate;

}

@Override

public RequestSpecificRetryHandler getRequestSpecificRetryHandler(

final S request, final IClientConfig requestConfig) {

if (this.okToRetryOnAllOperations) {

return new RequestSpecificRetryHandler(true, true, this.getRetryHandler(),

requestConfig);

}

if (!request.getContext().getMethod().equals(“GET”)) {

return new RequestSpecificRetryHandler(true, false, this.getRetryHandler(),

requestConfig);

}

else {

return new RequestSpecificRetryHandler(true, true, this.getRetryHandler(),

requestConfig);

}

}

protected boolean isSecure(final IClientConfig config) {

if(config != null) {

return RibbonProperties.from(config).isSecure(this.secure);

}

return this.secure;

}

@Override

protected void customizeLoadBalancerCommandBuilder(S request, IClientConfig config, LoadBalancerCommand.Builder builder) {

if (request.getLoadBalancerKey() != null) {

builder.withServerLocator(request.getLoadBalancerKey());

}

}

@Override

public ServiceInstance choose(String serviceId) {

Server server = this.getLoadBalancer().chooseServer(serviceId);

if (server != null) {

return new RibbonLoadBalancerClient.RibbonServer(serviceId, server);

}

return null;

}

public void validateServiceInstance(ServiceInstance serviceInstance) throws ClientException {

if (serviceInstance == null) {

throw new ClientException(“Load balancer does not have available server for client: ” + clientName);

} else if (serviceInstance.getHost() == null) {

throw new ClientException(“Invalid Server for: ” + serviceInstance.getServiceId() + ” null Host”);

}

}

}

• AbstractLoadBalancingClient继承了AbstractLoadBalancerAwareClient，实现了ServiceInstanceChooser接口
• 它有两个基本的实现，分别是用apache httpclient的RibbonLoadBalancingHttpClient，以及用okhttp的OkHttpLoadBalancingClient

RibbonLoadBalancingHttpClient

spring-cloud-netflix-ribbon-2.0.0.RELEASE-sources.jar!/org/springframework/cloud/netflix/ribbon/apache/RibbonLoadBalancingHttpClient.java

/**

* @author Christian Lohmann

* @author Ryan Baxter

* @author Tim Ysewyn

*/

// TODO: rename (ie new class that extends this in Dalston) to ApacheHttpLoadBalancingClient

public class RibbonLoadBalancingHttpClient extends

AbstractLoadBalancingClient {

public RibbonLoadBalancingHttpClient(IClientConfig config,

ServerIntrospector serverIntrospector) {

super(config, serverIntrospector);

}

public RibbonLoadBalancingHttpClient(CloseableHttpClient delegate,

IClientConfig config, ServerIntrospector serverIntrospector) {

super(delegate, config, serverIntrospector);

}

protected CloseableHttpClient createDelegate(IClientConfig config) {

RibbonProperties ribbon = RibbonProperties.from(config);

return HttpClientBuilder.create()

// already defaults to 0 in builder, so resetting to 0 won't hurt

.setMaxConnTotal(ribbon.maxTotalConnections(0))

// already defaults to 0 in builder, so resetting to 0 won't hurt

.setMaxConnPerRoute(ribbon.maxConnectionsPerHost(0))

.disableCookieManagement().useSystemProperties() // for proxy

.build();

}

@Override

public RibbonApacheHttpResponse execute(RibbonApacheHttpRequest request,

final IClientConfig configOverride) throws Exception {

IClientConfig config = configOverride != null ? configOverride : this.config;

RibbonProperties ribbon = RibbonProperties.from(config);

RequestConfig requestConfig = RequestConfig.custom()

.setConnectTimeout(ribbon.connectTimeout(this.connectTimeout))

.setSocketTimeout(ribbon.readTimeout(this.readTimeout))

.setRedirectsEnabled(ribbon.isFollowRedirects(this.followRedirects))

.build();

request = getSecureRequest(request, configOverride);

final HttpUriRequest httpUriRequest = request.toRequest(requestConfig);

final HttpResponse httpResponse = this.delegate.execute(httpUriRequest);

return new RibbonApacheHttpResponse(httpResponse, httpUriRequest.getURI());

}

@Override

public URI reconstructURIWithServer(Server server, URI original) {

URI uri = updateToSecureConnectionIfNeeded(original, this.config, this.serverIntrospector,

server);

return super.reconstructURIWithServer(server, uri);

}

@Override

public RequestSpecificRetryHandler getRequestSpecificRetryHandler(

RibbonApacheHttpRequest request, IClientConfig requestConfig) {

return new RequestSpecificRetryHandler(false, false, RetryHandler.DEFAULT,

requestConfig);

}

protected RibbonApacheHttpRequest getSecureRequest(RibbonApacheHttpRequest request, IClientConfig configOverride) {

if (isSecure(configOverride)) {

final URI secureUri = UriComponentsBuilder.fromUri(request.getUri())

.scheme(“https”).build(true).toUri();

return request.withNewUri(secureUri);

}

return request;

}

}

• 这个类被命名为ApacheHttpLoadBalancingClient更好一点
• execute方法将RibbonApacheHttpRequest转换为HttpUriRequest，而后用apache httpclient进行请求

OkHttpLoadBalancingClient

spring-cloud-netflix-ribbon-2.0.0.RELEASE-sources.jar!/org/springframework/cloud/netflix/ribbon/okhttp/OkHttpLoadBalancingClient.java

/**

* @author Spencer Gibb

* @author Ryan Baxter

* @author Tim Ysewyn

*/

public class OkHttpLoadBalancingClient

extends AbstractLoadBalancingClient {

public OkHttpLoadBalancingClient(IClientConfig config,

ServerIntrospector serverIntrospector) {

super(config, serverIntrospector);

}

public OkHttpLoadBalancingClient(OkHttpClient delegate, IClientConfig config,

ServerIntrospector serverIntrospector) {

super(delegate, config, serverIntrospector);

}

@Override

protected OkHttpClient createDelegate(IClientConfig config) {

return new OkHttpClient();

}

@Override

public OkHttpRibbonResponse execute(OkHttpRibbonRequest ribbonRequest,

final IClientConfig configOverride) throws Exception {

boolean secure = isSecure(configOverride);

if (secure) {

final URI secureUri = UriComponentsBuilder.fromUri(ribbonRequest.getUri())

.scheme(“https”).build().toUri();

ribbonRequest = ribbonRequest.withNewUri(secureUri);

}

OkHttpClient httpClient = getOkHttpClient(configOverride, secure);

final Request request = ribbonRequest.toRequest();

Response response = httpClient.newCall(request).execute();

return new OkHttpRibbonResponse(response, ribbonRequest.getUri());

}

OkHttpClient getOkHttpClient(IClientConfig configOverride, boolean secure) {

IClientConfig config = configOverride != null ? configOverride : this.config;

RibbonProperties ribbon = RibbonProperties.from(config);

OkHttpClient.Builder builder = this.delegate.newBuilder()

.connectTimeout(ribbon.connectTimeout(this.connectTimeout), TimeUnit.MILLISECONDS)

.readTimeout(ribbon.readTimeout(this.readTimeout), TimeUnit.MILLISECONDS)

.followRedirects(ribbon.isFollowRedirects(this.followRedirects));

if (secure) {

builder.followSslRedirects(ribbon.isFollowRedirects(this.followRedirects));

}

return builder.build();

}

@Override

public URI reconstructURIWithServer(Server server, URI original) {

URI uri = updateToSecureConnectionIfNeeded(original, this.config, this.serverIntrospector,

server);

return super.reconstructURIWithServer(server, uri);

}

}

• OkHttpLoadBalancingClient里头根据IClientConfig构建OkHttpClient，而后再将响应包装为OkHttpRibbonResponse

配置

HttpClientRibbonConfiguration

spring-cloud-netflix-ribbon-2.0.0.RELEASE-sources.jar!/org/springframework/cloud/netflix/ribbon/apache/HttpClientRibbonConfiguration.java

@Configuration

@ConditionalOnClass(name = “org.apache.http.client.HttpClient”)

@ConditionalOnProperty(name = “ribbon.httpclient.enabled”, matchIfMissing = true)

public class HttpClientRibbonConfiguration {

@RibbonClientName

private String name = “client”;

@Configuration

protected static class ApacheHttpClientConfiguration {

private final Timer connectionManagerTimer = new Timer(

“RibbonApacheHttpClientConfiguration.connectionManagerTimer”, true);

private CloseableHttpClient httpClient;

@Autowired(required = false)

private RegistryBuilder registryBuilder;

@Bean

@ConditionalOnMissingBean(HttpClientConnectionManager.class)

public HttpClientConnectionManager httpClientConnectionManager(

IClientConfig config,

ApacheHttpClientConnectionManagerFactory connectionManagerFactory) {

RibbonProperties ribbon = RibbonProperties.from(config);

int maxTotalConnections = ribbon.maxTotalConnections();

int maxConnectionsPerHost = ribbon.maxConnectionsPerHost();

int timerRepeat = ribbon.connectionCleanerRepeatInterval();

long timeToLive = ribbon.poolKeepAliveTime();

TimeUnit ttlUnit = ribbon.getPoolKeepAliveTimeUnits();

final HttpClientConnectionManager connectionManager = connectionManagerFactory

.newConnectionManager(false, maxTotalConnections,

maxConnectionsPerHost, timeToLive, ttlUnit, registryBuilder);

this.connectionManagerTimer.schedule(new TimerTask() {

@Override

public void run() {

connectionManager.closeExpiredConnections();

}

}, 30000, timerRepeat);

return connectionManager;

}

@Bean

@ConditionalOnMissingBean(CloseableHttpClient.class)

public CloseableHttpClient httpClient(ApacheHttpClientFactory httpClientFactory,

HttpClientConnectionManager connectionManager, IClientConfig config) {

RibbonProperties ribbon = RibbonProperties.from(config);

Boolean followRedirects = ribbon.isFollowRedirects();

Integer connectTimeout = ribbon.connectTimeout();

RequestConfig defaultRequestConfig = RequestConfig.custom()

.setConnectTimeout(connectTimeout)

.setRedirectsEnabled(followRedirects).build();

this.httpClient = httpClientFactory.createBuilder().

setDefaultRequestConfig(defaultRequestConfig).

setConnectionManager(connectionManager).build();

return httpClient;

}

@PreDestroy

public void destroy() throws Exception {

connectionManagerTimer.cancel();

if(httpClient != null) {

httpClient.close();

}

}

}

@Bean

@ConditionalOnMissingBean(AbstractLoadBalancerAwareClient.class)

@ConditionalOnMissingClass(value = “org.springframework.retry.support.RetryTemplate”)

public RibbonLoadBalancingHttpClient ribbonLoadBalancingHttpClient(

IClientConfig config, ServerIntrospector serverIntrospector,

ILoadBalancer loadBalancer, RetryHandler retryHandler, CloseableHttpClient httpClient) {

RibbonLoadBalancingHttpClient client = new RibbonLoadBalancingHttpClient(httpClient, config, serverIntrospector);

client.setLoadBalancer(loadBalancer);

client.setRetryHandler(retryHandler);

Monitors.registerObject(“Client_” + this.name, client);

return client;

}

@Bean

@ConditionalOnMissingBean(AbstractLoadBalancerAwareClient.class)

@ConditionalOnClass(name = “org.springframework.retry.support.RetryTemplate”)

public RetryableRibbonLoadBalancingHttpClient retryableRibbonLoadBalancingHttpClient(

IClientConfig config, ServerIntrospector serverIntrospector,

ILoadBalancer loadBalancer, RetryHandler retryHandler,

LoadBalancedRetryFactory loadBalancedRetryFactory, CloseableHttpClient httpClient) {

RetryableRibbonLoadBalancingHttpClient client = new RetryableRibbonLoadBalancingHttpClient(

httpClient, config, serverIntrospector, loadBalancedRetryFactory);

client.setLoadBalancer(loadBalancer);

client.setRetryHandler(retryHandler);

Monitors.registerObject(“Client_” + this.name, client);

return client;

}

}

• 配置的是RibbonLoadBalancingHttpClient，假如有retry，则是RetryableRibbonLoadBalancingHttpClient

OkHttpRibbonConfiguration

spring-cloud-netflix-ribbon-2.0.0.RELEASE-sources.jar!/org/springframework/cloud/netflix/ribbon/okhttp/OkHttpRibbonConfiguration.java

@Configuration

@ConditionalOnProperty(“ribbon.okhttp.enabled”)

@ConditionalOnClass(name = “okhttp3.OkHttpClient”)

public class OkHttpRibbonConfiguration {

@RibbonClientName

private String name = “client”;

@Configuration

protected static class OkHttpClientConfiguration {

private OkHttpClient httpClient;

@Bean

@ConditionalOnMissingBean(ConnectionPool.class)

public ConnectionPool httpClientConnectionPool(IClientConfig config,

OkHttpClientConnectionPoolFactory connectionPoolFactory) {

RibbonProperties ribbon = RibbonProperties.from(config);

int maxTotalConnections = ribbon.maxTotalConnections();

long timeToLive = ribbon.poolKeepAliveTime();

TimeUnit ttlUnit = ribbon.getPoolKeepAliveTimeUnits();

return connectionPoolFactory.create(maxTotalConnections, timeToLive, ttlUnit);

}

@Bean

@ConditionalOnMissingBean(OkHttpClient.class)

public OkHttpClient client(OkHttpClientFactory httpClientFactory,

ConnectionPool connectionPool, IClientConfig config) {

RibbonProperties ribbon = RibbonProperties.from(config);

this.httpClient = httpClientFactory.createBuilder(false)

.connectTimeout(ribbon.connectTimeout(), TimeUnit.MILLISECONDS)

.readTimeout(ribbon.readTimeout(), TimeUnit.MILLISECONDS)

.followRedirects(ribbon.isFollowRedirects())

.connectionPool(connectionPool)

.build();

return this.httpClient;

}

@PreDestroy

public void destroy() {

if(httpClient != null) {

httpClient.dispatcher().executorService().shutdown();

httpClient.connectionPool().evictAll();

}

}

}

@Bean

@ConditionalOnMissingBean(AbstractLoadBalancerAwareClient.class)

@ConditionalOnClass(name = “org.springframework.retry.support.RetryTemplate”)

public RetryableOkHttpLoadBalancingClient retryableOkHttpLoadBalancingClient(

IClientConfig config,

ServerIntrospector serverIntrospector,

ILoadBalancer loadBalancer,

RetryHandler retryHandler,

LoadBalancedRetryFactory loadBalancedRetryFactory,

OkHttpClient delegate) {

RetryableOkHttpLoadBalancingClient client = new RetryableOkHttpLoadBalancingClient(delegate, config,

serverIntrospector, loadBalancedRetryFactory);

client.setLoadBalancer(loadBalancer);

client.setRetryHandler(retryHandler);

Monitors.registerObject(“Client_” + this.name, client);

return client;

}

@Bean

@ConditionalOnMissingBean(AbstractLoadBalancerAwareClient.class)

@ConditionalOnMissingClass(value = “org.springframework.retry.support.RetryTemplate”)

public OkHttpLoadBalancingClient okHttpLoadBalancingClient(

IClientConfig config,

ServerIntrospector serverIntrospector, ILoadBalancer loadBalancer,

RetryHandler retryHandler, OkHttpClient delegate) {

OkHttpLoadBalancingClient client = new OkHttpLoadBalancingClient(delegate, config,

serverIntrospector);

client.setLoadBalancer(loadBalancer);

client.setRetryHandler(retryHandler);

Monitors.registerObject(“Client_” + this.name, client);

return client;

}

}

• 配置的是OkHttpLoadBalancingClient，假如有retry，则是RetryableOkHttpLoadBalancingClient

小结

AbstractLoadBalancingClient有appche httpclient以及okhttp两类实现，分别是RibbonLoadBalancingHttpClient及OkHttpLoadBalancingClient；每种实现都有retry相关的实现，分别是RetryableRibbonLoadBalancingHttpClient、RetryableOkHttpLoadBalancingClient。

doc

• 16. Client Side Load Balancer: Ribbon