[享学Netflix] 五十四、Ribbon启动连接操作：IPrimeConnection检测Server是否能够提供服务

public interface IPrimeConnection extends IClientConfigAware {
	// 子类应该实现协议特定的连接操作到服务器。
	// server：待连接的服务器
	// uriPath：进行连接时使用的uri。如/api/v1/ping
    public boolean connect(Server server, String uriPath) throws Exception;
}

public class HttpPrimeConnection implements IPrimeConnection {

	// 它是个HC，继承自Apache的DefaultHttpClient
	private NFHttpClient client;

	// client名称是：clientName-PrimeConnsClient
	// 重点是：链接超时时间是2s。若2s还没连接上就代表链接失败
    @Override
    public void initWithNiwsConfig(IClientConfig niwsClientConfig) {
        client = NFHttpClientFactory.getNamedNFHttpClient(niwsClientConfig.getClientName() + "-PrimeConnsClient", false); 
        HttpConnectionParams.setConnectionTimeout(client.getParams(), 2000);        
    }
    
	//发送一个get请求，只要connect连接上了，就表示成功了（状态码无所谓，因为只是测试connect连接性）
    @Override
    public boolean connect(Server server, String primeConnectionsURIPath) throws Exception {
        String url = "http://" + server.getHostPort() + primeConnectionsURIPath;
        logger.debug("Trying URL: {}", url);
        HttpUriRequest get = new HttpGet(url);
        HttpResponse response = null;
        try {
            response = client.execute(get);
            if (logger.isDebugEnabled() && response.getStatusLine() != null) {
                logger.debug("Response code:" + response.getStatusLine().getStatusCode());
            }
        } finally {
           get.abort();
        }
        return true;
    }

}

public class PrimeConnections {

	String primeConnectionsURIPath = "/";
    private ExecutorService executorService;
    private int maxExecutorThreads = 5;
    private long executorThreadTimeout = 30000;
    private String name = "default";
    private float primeRatio = 1.0f;
    int maxRetries = 9;
    long maxTotalTimeToPrimeConnections = 30 * 1000;
    long totalTimeTaken = 0;
    private boolean aSync = true;
        
    // servo里用于监控的类，忽略
    Counter totalCounter;
    Counter successCounter;
    Timer initialPrimeTimer;

    private IPrimeConnection connector;
    private PrimeConnectionEndStats stats;
	
}

executorService = new ThreadPoolExecutor(
	1, // 核心线程数只有1个
	maxExecutorThreads, // 默认是5
	executorThreadTimeout, // 默认30s
	TimeUnit.MILLISECONDS,
	new LinkedBlockingQueue<Runnable>(), // 因为Server不会超级多，所以用无界队列也没关系
	new ASyncPrimeConnectionsThreadFactory(name));

public void primeConnections(List<Server> servers) { ... }
public List<Future<Boolean>> primeConnectionsAsync(final List<Server> servers, final PrimeConnectionListener listener) { ... }

// 这两个简单，不用做单独讲解和梳理
public PrimeConnectionEndStats getEndStats() { ... }
public void shutdown() { ... }

PrimeConnections：

	public List<Future<Boolean>> primeConnectionsAsync(final List<Server> servers, final PrimeConnectionListener listener) {
		...
		List<Future<Boolean>> ftList = new ArrayList<>();
		for (Server s : allServers) {
			s.setReadyToServe(false);
			
			// 默认是开启了异步方式的，任务交给线程池处理
			if (aSync) {
				ftList.add(executorService.submit(() -> connectToServer(s, listener)));
			} else {
				connectToServer(s, listener);
			}	
		}
	}

	// 使用IPrimeConnection#connect完成连接动作
	// 对于http的connect来说，超时是2s，所以此方法最长hold 18秒
	private Boolean connectToServer(final Server s, final PrimeConnectionListener listener) {
		boolean success = false;
		do {
			success = connector.connect(s, primeConnectionsURIPath);
			successCounter.increment();
			lastException = null;
			//连接成功后立马停止
			break;
		} while (!success && (tryNum <= maxRetries));
			
		if (listener != null) {
			listener.primeCompleted(s, lastException);
		}
		return success;
	}

	// 以递增的方式，每次增加重试时的睡眠时长，这是非常合理的。如：
	// 200ms, 400 ms, 800ms, 1600ms etc.
    private void sleepBeforeRetry(int tryNum) {
        try {
            int sleep = (tryNum + 1) * 100;
            logger.debug("Sleeping for " + sleep + "ms ...");
            Thread.sleep(sleep);
        } catch (InterruptedException ex) {
        }
    }

BaseLoadBalancer：

	public void setServersList(List lsrv) {
		...
		if (isEnablePrimingConnections()) {
			...
            if (primeConnections != null) {
                primeConnections.primeConnectionsAsync(newServers, this);
            }
		}
		...
	}

BaseLoadBalancer：

    @Override
    public void primeCompleted(Server s, Throwable lastException) {
        s.setReadyToServe(true);
    }

PrimeConnections：

	public void primeConnections(List<Server> servers) {

		// 准备发送链接之前，先全部置为false（这种操作是很危险的，所以请保证连通性）
        for (Server server: servers) {
            server.setReadyToServe(false);
        }		
        // 默认100%执行完了（并不是通过）才会放行主线程
        // 此处使用CountDownLatch 闭锁来控制，异步 -> 同步阻塞
        int totalCount = (int) (servers.size() * primeRatio); 
        CountDownLatch latch = new CountDownLatch(totalCount);
        // 只要木有抛出异常（证明联通了），那就s.setReadyToServe(true);
        primeConnectionsAsync(servers, new PrimeConnectionListener()  {            
            @Override
            public void primeCompleted(Server s, Throwable lastException) {
                if (lastException == null) {
                    successCount.incrementAndGet();
                    s.setReadyToServe(true);
                } else {
                    failureCount.incrementAndGet();
                }
                latch.countDown();
            }
        }); 
		...
		// 如果机器太多，也不会一直等待你，最大等待时间，默认是30s
		// 当然可以通过`MaxTotalTimeToPrimeConnections`这个key来设定
		latch.await(maxTotalTimeToPrimeConnections, TimeUnit.MILLISECONDS);
		...
		// 记录链接的结果，后面可以通过get方法获取到
		stats = new PrimeConnectionEndStats(totalCount, successCount.get(), failureCount.get(), stopWatch.getDuration(TimeUnit.MILLISECONDS));
		// 打印一下结果。结果为info级别：Priming Connections not fully successful
		// 详细日志是debug级别
		printStats(stats);
	}