上一篇我们了解到Ribbon主要由如下几个组件组成：
- 所有Ribbon负载均衡器需要实现的接口IClient
- 服务实例列表维护机制实现的接口ServerList
- 负载均衡数据记录LoadBalancerStats
- 负责选取Server的接口ILoadBalancer
- 负载均衡选取规则实现的接口IRule
- 检查实例是否存活实现的接口IPing
- 服务实例列表更新机制实现的接口ServerListUpdater
- 服务实例列表过滤机制ServerListFilter
我们会逐个分析
1. 所有Ribbon负载均衡器需要实现的接口IClient
对于这个IClient，之前我们说到执行器逻辑，例如重试还有异常处理，都在这里处理。我们看他的默认抽象类实现AbstractLoadBalancerAwareClient：
AbstractLoadBalancerAwareClient.java
public T executeWithLoadBalancer(final S request, final IClientConfig requestConfig) throws ClientException { //获取重试处理器，这个由其他实现类动态实现 RequestSpecificRetryHandler handler = getRequestSpecificRetryHandler(request, requestConfig); //构造LoadBalancerCommand，RxJava风格 LoadBalancerCommand<T> command = LoadBalancerCommand.<T>builder() .withLoadBalancerContext(this) .withRetryHandler(handler) .withLoadBalancerURI(request.getUri()) .build(); try { return command.submit( new ServerOperation<T>() { @Override public Observable<T> call(Server server) { //修改原始url为实际的url URI finalUri = reconstructURIWithServer(server, request.getUri()); S requestForServer = (S) request.replaceUri(finalUri); try { //执行请求 return Observable.just(AbstractLoadBalancerAwareClient.this.execute(requestForServer, requestConfig)); } catch (Exception e) { return Observable.error(e); } } }) .toBlocking() .single(); } catch (Exception e) { Throwable t = e.getCause(); if (t instanceof ClientException) { throw (ClientException) t; } else { throw new ClientException(e); } } } public abstract RequestSpecificRetryHandler getRequestSpecificRetryHandler(S request, IClientConfig requestConfig);
这个构造的LoadBalancerCommand是一个RxJava风格的，它包含了重试和异常处理机制：
LoadBalancerCommand.java
//返回一个只包含一个Server的Observable，但是每次从负载均衡器中获取一个 private Observable<Server> selectServer() { return Observable.create(new OnSubscribe<Server>() { @Override public void call(Subscriber<? super Server> next) { try { Server server = loadBalancerContext.getServerFromLoadBalancer(loadBalancerURI, loadBalancerKey); next.onNext(server); next.onCompleted(); } catch (Exception e) { next.onError(e); } } }); } public Observable<T> submit(final ServerOperation<T> operation) { final ExecutionInfoContext context = new ExecutionInfoContext(); if (listenerInvoker != null) { try { listenerInvoker.onExecutionStart(); } catch (AbortExecutionException e) { return Observable.error(e); } } //获取在每个服务实例重试的的次数 final int maxRetrysSame = retryHandler.getMaxRetriesOnSameServer(); //最多尝试几个服务实例 final int maxRetrysNext = retryHandler.getMaxRetriesOnNextServer(); //对于每个服务实例的调用逻辑 //默认field server是null，通过selectServer()方法获取一个Server Observable<T> o = (server == null ? selectServer() : Observable.just(server)) .concatMap(new Func1<Server, Observable<T>>() { @Override //对于每个Server，按顺序映射为对于每个Server包含重试逻辑的请求调用 public Observable<T> call(Server server) { //设置上下文 context.setServer(server); final ServerStats stats = loadBalancerContext.getServerStats(server); //每个Server包含重试逻辑的请求调用 Observable<T> o = Observable .just(server) .concatMap(new Func1<Server, Observable<T>>() { @Override public Observable<T> call(final Server server) { context.incAttemptCount(); //增加Server正在处理的请求计数 loadBalancerContext.noteOpenConnection(stats); //监听器 if (listenerInvoker != null) { try { listenerInvoker.onStartWithServer(context.toExecutionInfo()); } catch (AbortExecutionException e) { return Observable.error(e); } } //计时器 final Stopwatch tracer = loadBalancerContext.getExecuteTracer().start(); //operation.call(server)就是刚刚分析的AbstractLoadBalancerAwareClient传过来的ServerOperation，就是直接对这个Server调用请求 //doOnEach的操作就是记录请求前后的一些数据用于负载均衡数据统计 return operation.call(server).doOnEach(new Observer<T>() { private T entity; @Override public void onCompleted() { //记录请求完成 recordStats(tracer, stats, entity, null); } @Override public void onError(Throwable e) { //记录请求结束 recordStats(tracer, stats, null, e); logger.debug("Got error {} when executed on server {}", e, server); //发生了错误，通知listener if (listenerInvoker != null) { listenerInvoker.onExceptionWithServer(e, context.toExecutionInfo()); } } @Override public void onNext(T entity) { //因为只有调用请求成功只有一个结果（只有一个请求）， 这里的entity就是结果，只要收到结果就代表请求成功 this.entity = entity; if (listenerInvoker != null) { listenerInvoker.onExecutionSuccess(entity, context.toExecutionInfo()); } } private void recordStats(Stopwatch tracer, ServerStats stats, Object entity, Throwable exception) { tracer.stop(); loadBalancerContext.noteRequestCompletion(stats, entity, exception, tracer.getDuration(TimeUnit.MILLISECONDS), retryHandler); } }); } }); if (maxRetrysSame > 0) //是否retry o = o.retry(retryPolicy(maxRetrysSame, true)); return o; } }); if (maxRetrysNext > 0 && server == null) //是否retry，如果retry回调用selectServer()返回下一个Server o = o.retry(retryPolicy(maxRetrysNext, false)); //异常处理 return o.onErrorResumeNext(new Func1<Throwable, Observable<T>>() { @Override public Observable<T> call(Throwable e) { if (context.getAttemptCount() > 0) { //如果超过重试次数，则抛异常 if (maxRetrysNext > 0 && context.getServerAttemptCount() == (maxRetrysNext + 1)) { e = new ClientException(ClientException.ErrorType.NUMBEROF_RETRIES_NEXTSERVER_EXCEEDED, "Number of retries on next server exceeded max " + maxRetrysNext + " retries, while making a call for: " + context.getServer(), e); } else if (maxRetrysSame > 0 && context.getAttemptCount() == (maxRetrysSame + 1)) { e = new ClientException(ClientException.ErrorType.NUMBEROF_RETRIES_EXEEDED, "Number of retries exceeded max " + maxRetrysSame + " retries, while making a call for: " + context.getServer(), e); } } if (listenerInvoker != null) { listenerInvoker.onExecutionFailed(e, context.toFinalExecutionInfo()); } return Observable.error(e); } }); }
2. 服务实例列表维护机制实现的接口ServerList
AbstractServerList.java
其实这个抽象类一是在实现ServerList接口的同时，实现了IClientConfigAware这个接口，代表是可配置的。
同时，提供了一个生成默认ServerListFilter（这个Filter的实现类是由NIWSServerListFilterClassName这个配置决定，默认是ZoneAffinityServerListFilter）的方法
public abstract class AbstractServerList<T extends Server> implements ServerList<T>, IClientConfigAware { public AbstractServerListFilter<T> getFilterImpl(IClientConfig niwsClientConfig) throws ClientException{ try { String niwsServerListFilterClassName = niwsClientConfig .getProperty( CommonClientConfigKey.NIWSServerListFilterClassName, ZoneAffinityServerListFilter.class.getName()) .toString(); AbstractServerListFilter<T> abstractNIWSServerListFilter = (AbstractServerListFilter<T>) ClientFactory.instantiateInstanceWithClientConfig(niwsServerListFilterClassName, niwsClientConfig); return abstractNIWSServerListFilter; } catch (Throwable e) { throw new ClientException( ClientException.ErrorType.CONFIGURATION, "Unable to get an instance of CommonClientConfigKey.NIWSServerListFilterClassName. Configured class:" + niwsClientConfig .getProperty(CommonClientConfigKey.NIWSServerListFilterClassName), e); } } }
ConfigurationBasedServerList.java
这个是默认的实现，如果没有特殊配置，ServerList的实现类就是ConfigurationBasedServerList；这个实际上就是从配置中读取ServerList，这个配置可以是动态配置，例如是Archaius
public class ConfigurationBasedServerList extends AbstractServerList<Server> { private IClientConfig clientConfig; @Override public List<Server> getInitialListOfServers() { return getUpdatedListOfServers(); } @Override public List<Server> getUpdatedListOfServers() { String listOfServers = clientConfig.get(CommonClientConfigKey.ListOfServers); return derive(listOfServers); } @Override public void initWithNiwsConfig(IClientConfig clientConfig) { this.clientConfig = clientConfig; } //可以看出这个配置就是以逗号分隔的字符串 private List<Server> derive(String value) { List<Server> list = Lists.newArrayList(); if (!Strings.isNullOrEmpty(value)) { for (String s: value.split(",")) { list.add(new Server(s.trim())); } } return list; } }
DiscoveryEnabledNIWSServerList.java
这个就是从Eureka上面获取Server列表的类，构造的时候需要传入相关配置以及最重要的EurekaClient的Provider来获取合适的EurekaClient以便于获取Server列表。
实现ServerList接口的方法都是基于obtainServersViaDiscovery这个方法：
@Override public List<DiscoveryEnabledServer> getInitialListOfServers(){ return obtainServersViaDiscovery(); } @Override public List<DiscoveryEnabledServer> getUpdatedListOfServers(){ return obtainServersViaDiscovery(); } private List<DiscoveryEnabledServer> obtainServersViaDiscovery() { List<DiscoveryEnabledServer> serverList = new ArrayList<DiscoveryEnabledServer>(); //如果EurekaClient没有被初始化，则日志报警并返回空的列表 if (eurekaClientProvider == null || eurekaClientProvider.get() == null) { logger.warn("EurekaClient has not been initialized yet, returning an empty list"); return new ArrayList<DiscoveryEnabledServer>(); } EurekaClient eurekaClient = eurekaClientProvider.get(); //这里的vipAddresses其实就是微服务名称的各种形式，但是注意，它们代表的是同一个微服务 if (vipAddresses!=null){ for (String vipAddress : vipAddresses.split(",")) { // if targetRegion is null, it will be interpreted as the same region of client List<InstanceInfo> listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, isSecure, targetRegion); for (InstanceInfo ii : listOfInstanceInfo) { if (ii.getStatus().equals(InstanceStatus.UP)) { //是否覆盖port if(shouldUseOverridePort){ if(logger.isDebugEnabled()){ logger.debug("Overriding port on client name: " + clientName + " to " + overridePort); } //这里复制一份是因为不希望其他的地方修改原有的实例信息 InstanceInfo copy = new InstanceInfo(ii); if(isSecure){ ii = new InstanceInfo.Builder(copy).setSecurePort(overridePort).build(); }else{ ii = new InstanceInfo.Builder(copy).setPort(overridePort).build(); } } DiscoveryEnabledServer des = new DiscoveryEnabledServer(ii, isSecure, shouldUseIpAddr); des.setZone(DiscoveryClient.getZone(ii)); serverList.add(des); } } //如果有一个vipAddress有服务列表，我们就不用获取剩余的了 if (serverList.size()>0 && prioritizeVipAddressBasedServers){ break; } } } return serverList; }
到这里我们可以看出，Ribbon和Eureka的配合其实就是Ribbon从Eureka中利用微服务名称获取Server列表；那么这个列表是如何更新的呢，在Eureka的章节我们提到过，Ribbon定时从EurekaClient获取服务实例列表更新，这就涉及到了下一个我们要讲到的Ribbon元素 - 服务实例列表更新机制实现的接口ServerListUpdater
