ribbon源码之客户端

　　客户端模块的核心功能是提供统一的用户请求操作接口。

接口定义

　　客户端模块的核心是IClient接口，定义了客户端网络请求的方法。

public interface IClient<S extends ClientRequest, T extends IResponse> {

    public T execute(S request, IClientConfig requestConfig) throws Exception;

}

　　ClientRequest为客户端定义的请求体，存储了请求uri、loadbalancer的key，是否重试、配置。

public class ClientRequest implements Cloneable {

    protected URI uri;

    protected Object loadBalancerKey = null;

    protected Boolean isRetriable = null;

    protected IClientConfig overrideConfig;

}

　　IResponse为客户端定义的响应内容的接口。

public interface IResponse extends Closeable

{

   public Object getPayload() throws ClientException;

   public boolean hasPayload();

   public boolean isSuccess();

   public URI getRequestedURI();

   public Map<String, ?> getHeaders();

}

　　IClientConfigAware定义了需要使用IClientConfig初始化IClient的方法。

public interface IClientConfigAware {

    public abstract void initWithNiwsConfig(IClientConfig clientConfig);

}

　　ribbon基于http请求，相关类和接口，HttpRequest为http请求；HttpResponse为http请求返回结果；Restclient是基于jesery的IClient实现。

类图

客户端工厂类

　　客户端模块提供了一个客户端工厂类（ClientFactory）用于通过配置文件来创建IClient实例和负载均衡器（ILoadBalancer）实例。

public static synchronized IClient getNamedClient(String name) {//根据配置获取iclient实例，默认使用DefaultClientConfigImpl配置类。

        return getNamedClient(name, DefaultClientConfigImpl.class);

    }

public static synchronized IClient getNamedClient(String name, Class<? extends IClientConfig> configClass) {
   ...
        return createNamedClient(name, configClass);
   ...
}

public static synchronized IClient createNamedClient(String name, Class<? extends IClientConfig> configClass) throws ClientException {
   IClientConfig config = getNamedConfig(name, configClass);//实例化配置类
    return registerClientFromProperties(name, config);//通过配置文件创建iclient
}

public static synchronized ILoadBalancer getNamedLoadBalancer(String name) {
   return getNamedLoadBalancer(name, DefaultClientConfigImpl.class);
}
public static synchronized ILoadBalancer getNamedLoadBalancer(String name, Class<? extends IClientConfig> configClass) {
   ...
            lb = registerNamedLoadBalancerFromProperties(name, configClass);
   ...
}

public static synchronized IClient<?, ?> registerClientFromProperties(String restClientName, IClientConfig clientConfig) throws ClientException {

        IClient<?, ?> client = null;

        ILoadBalancer loadBalancer = null;

        ...

            String clientClassName = (String) clientConfig.getProperty(CommonClientConfigKey.ClientClassName);//通过配置文件获取client的实现类

            client = (IClient<?, ?>) instantiateInstanceWithClientConfig(clientClassName, clientConfig); //通过配置文件创建client实例

            boolean initializeNFLoadBalancer = Boolean.parseBoolean(clientConfig.getProperty(

                    CommonClientConfigKey.InitializeNFLoadBalancer, DefaultClientConfigImpl.DEFAULT_ENABLE_LOADBALANCER).toString());

            if (initializeNFLoadBalancer) {//如果需要初始化负载均衡器，则通过配置文件创建一个负载均衡器

                loadBalancer  = registerNamedLoadBalancerFromclientConfig(restClientName, clientConfig);

            }

            if (client instanceof AbstractLoadBalancerAwareClient) {//如果client实现AbstractLoadBalancerAwareClient，则注入负载均衡器

                ((AbstractLoadBalancerAwareClient) client).setLoadBalancer(loadBalancer);

            }

        ...return client;

    }

public static ILoadBalancer registerNamedLoadBalancerFromclientConfig(String name, IClientConfig clientConfig) throws ClientException {
   ...
   ILoadBalancer lb = null;
   ...
        String loadBalancerClassName = (String) clientConfig.getProperty(CommonClientConfigKey.NFLoadBalancerClassName);//
        lb = (ILoadBalancer) ClientFactory.instantiateInstanceWithClientConfig(loadBalancerClassName, clientConfig);                                    
        ...
        return lb;
   ...      
}
//初始化指定的class类
public static Object instantiateInstanceWithClientConfig(String className, IClientConfig clientConfig)

          throws InstantiationException, IllegalAccessException, ClassNotFoundException {
       Class clazz = Class.forName(className);
       if (IClientConfigAware.class.isAssignableFrom(clazz)) {//如果指定的iclient实现了IClientConfigAware，ClientFactory在创建时会使用IClientConfig进行初始化。
          IClientConfigAware obj = (IClientConfigAware) clazz.newInstance();
          obj.initWithNiwsConfig(clientConfig);
          return obj;
       } else {
          try {
              if (clazz.getConstructor(IClientConfig.class) != null) {
                 return clazz.getConstructor(IClientConfig.class).newInstance(clientConfig);
              }
          } catch (Throwable e) { // NOPMD             
          }         
       }
       return clazz.newInstance();
   }

　　使用客户端工厂类（ClientFactory）涉及的配置：

属性	实现	默认值
clientname.ribbon.ClientClassName	client使用的IClient实现类	com.netflix.niws.client.http.RestClient
clientname.ribbon.InitializeNFLoadBalancer	是否初始化负载均衡器	true
clientname.ribbon.NFLoadBalancerClassName	负载均衡器的实现类	com.netflix.loadbalancer.ZoneAwareLoadBalancer

类图

客户端实现类

　　AbstractLoadBalancerAwareClient实现了通过负载均衡器进行请求调用。LoadBalancerCommand对负载均衡器操作进行了模版，对请求调用提供了回调函数。

public T executeWithLoadBalancer(final S request, final IClientConfig requestConfig) throws ClientException {

        LoadBalancerCommand<T> command = buildLoadBalancerCommand(request, requestConfig);

　　　　 ...return command.submit(

                new ServerOperation<T>() {

                    @Override

                    public Observable<T> call(Server server) {

                        URI finalUri = reconstructURIWithServer(server, request.getUri());

                        S requestForServer = (S) request.replaceUri(finalUri);//设置最终的调用uri。

                        try {

                            return Observable.just(AbstractLoadBalancerAwareClient.this.execute(requestForServer, requestConfig));

                        } //调用execute方法执行请求调用。

                        catch (Exception e) {

                            return Observable.error(e);

                        }

                    }

                })

                .toBlocking()

                .single();

        ...

    }

　　LoadBalancerCommand调用了负载均衡器获得了一个server，然后调用回调函数执行请求。此外还提供了各个关键节点的监听器和异常重试机制。

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();

        Observable<T> o =

                (server == null ? selectServer() : Observable.just(server))//调用负载均衡器获得目标server

                .concatMap(new Func1<Server, Observable<T>>() {

                    ...return operation.call(server).doOnEach(new Observer<T>() {

                                            ....

                                        });

                    ...

                        if (maxRetrysSame > 0)

                            o = o.retry(retryPolicy(maxRetrysSame, true));

                        return o;

                    }

                });

        if (maxRetrysNext > 0 && server == null)

            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);

            }

        });

    }

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);

                }

            }

        });

    }

　　子类 HttpRequest用于http请求，内部定义了http请求的各个内容，并且使用了builder模式。

public class HttpRequest extends ClientRequest {protected CaseInsensitiveMultiMap httpHeaders = new CaseInsensitiveMultiMap();//head参数

    protected Multimap<String, String> queryParams = ArrayListMultimap.create();//query参数

    private Object entity;／／消息体

    protected Verb verb;//http请求的method：post get head delete等。默认get

}

类图

巴特西

ribbon源码之客户端

最新文章

热门文章