聊聊spring cloud的AbstractLoadBalancingClient

時間 2019-11-06

標籤聊聊 spring cloud abstractloadbalancingclient 欄目 Spring 简体版

原文原文鏈接

序

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

AbstractLoadBalancingClient

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

/**
 * @author Spencer Gibb
 */
public abstract class AbstractLoadBalancingClient<S extends ContextAwareRequest, T extends IResponse, D> extends
		AbstractLoadBalancerAwareClient<S, T> 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<T> 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.javaspring

/**
 * @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<RibbonApacheHttpRequest, RibbonApacheHttpResponse, CloseableHttpClient> {

	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.javaapache

/**
 * @author Spencer Gibb
 * @author Ryan Baxter
 * @author Tim Ysewyn
 */
public class OkHttpLoadBalancingClient
		extends AbstractLoadBalancingClient<OkHttpRibbonRequest, OkHttpRibbonResponse, OkHttpClient> {

	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.javaapp

@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.javaide

@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。ui