ribbon源码之client(代码片段)

public interface IClient<S extends ClientRequest, T extends IResponse> 
    public T execute(S request, IClientConfig requestConfig) throws Exception; 

public class ClientRequest implements Cloneable 
    protected URI uri;
    protected Object loadBalancerKey = null;
    protected Boolean isRetriable = null;
    protected IClientConfig overrideConfig;

public interface IResponse extends Closeable

   public Object getPayload() throws ClientException;
   public boolean hasPayload();
   public boolean isSuccess();
   public URI getRequestedURI();
   public Map<String, ?> getHeaders();   

public interface IClientConfigAware 
    public abstract void initWithNiwsConfig(IClientConfig clientConfig);

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

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

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

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