[享学Netflix] 五十、Ribbon的LoadBalancer五大组件之：ServerListUpdater

// Server列表更新器。
public interface ServerListUpdater {
	// 内部接口：函数式接口 实际上执行服务器列表更新的接口 也就是实际的动作
	// 一般使用匿名内部类的形式实现
	public interface UpdateAction {
        void doUpdate();
    }

	// 使用给定的更新操作启动serverList更新程序这个调用应该是幂等的
	void start(UpdateAction updateAction);
	// 停止服务器列表更新程序。这个调用应该是幂等的
	void stop();

	// ============下面是一些获取执行过程中的信息方法==============
	// 最后更新的时间Date的String表示形式
	String getLastUpdate();
	// 自上次更新以来已经过的ms数
	long getDurationSinceLastUpdateMs();
	//错过更新周期的数量（如果有的话）
	int getNumberMissedCycles();
	// 使用的线程数
	int getCoreThreads();
}

public class PollingServerListUpdater implements ServerListUpdater {

    private static long LISTOFSERVERS_CACHE_UPDATE_DELAY = 1000;
    private static int LISTOFSERVERS_CACHE_REPEAT_INTERVAL = 30 * 1000;

	private final AtomicBoolean isActive = new AtomicBoolean(false);
	private volatile long lastUpdated = System.currentTimeMillis();
    private final long initialDelayMs;
    private final long refreshIntervalMs;
    
	// 继承自Futrue，在Futrue的基础上增加getDelay(TimeUnit unit)方法：还有多久执行任务
	// ScheduledExecutorService提交任务时返回它
	// ScheduledThreadPoolExecutor是带有线程池功能的执行器，实现了接口ScheduledExecutorService
	private volatile ScheduledFuture<?> scheduledFuture;
}

PollingServerListUpdater.LazyHolder：

	// 重点：core核心数是可**动态**可变的
	private final static String CORE_THREAD = "DynamicServerListLoadBalancer.ThreadPoolSize";
	private final static DynamicIntProperty poolSizeProp = new DynamicIntProperty(CORE_THREAD, 2);
	static ScheduledThreadPoolExecutor _serverListRefreshExecutor = null;

	// 使用静态代码块完成初始化
	static {
		// 注册一个callback以便能动态更新core核心数
		int coreSize = poolSizeProp.get();
        poolSizeProp.addCallback(new Runnable() {
            @Override
            public void run() {
                _serverListRefreshExecutor.setCorePoolSize(poolSizeProp.get());
            }
        });
        _serverListRefreshExecutor = new ScheduledThreadPoolExecutor(coreSize, factory);
        ...
	}

PollingServerListUpdater：

	// 构造器，为initialDelayMs/refreshIntervalMs两个参数赋值
    public PollingServerListUpdater() {
        this(LISTOFSERVERS_CACHE_UPDATE_DELAY, LISTOFSERVERS_CACHE_REPEAT_INTERVAL);
    }
    // 从config里面拿值。对应的key是ServerListRefreshInterval
    public PollingServerListUpdater(IClientConfig clientConfig) {
        this(LISTOFSERVERS_CACHE_UPDATE_DELAY, getRefreshIntervalMs(clientConfig));
    }
    public PollingServerListUpdater(final long initialDelayMs, final long refreshIntervalMs) {
        this.initialDelayMs = initialDelayMs;
        this.refreshIntervalMs = refreshIntervalMs;
    }

	// 启动
    @Override
    public synchronized void start(final UpdateAction updateAction) {
    	// 保证原子性。如果已经启动了就啥都不做
    	if (isActive.compareAndSet(false, true)) {
			//定时任务每次执行的Task
    		Runnable wrapperRunnable = () -> {
				if (!isActive.get()) {
					if (scheduledFuture != null) {
						scheduledFuture.cancel(true);
						return;
					}
					// 每次执行更新操作时，记录下时间戳
					updateAction.doUpdate();
					lastUpdated = System.currentTimeMillis();
				}

				// 启动任务 默认30s执行一次
	            scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay(
	                    wrapperRunnable, initialDelayMs,
	                    refreshIntervalMs, TimeUnit.MILLISECONDS
    		};
    	} 
    }

	// 停止任务
    @Override
    public synchronized void stop() {
    	scheduledFuture.cancel(true);
    }

	// 为何返回String？返回Date不香吗？
    @Override
    public String getLastUpdate() {
        return new Date(lastUpdated).toString();
    }
    // 距离上一次update更新已过去这么长时间了
    @Override
    public long getDurationSinceLastUpdateMs() {
        return System.currentTimeMillis() - lastUpdated;
    }

	// 因为coreSize是动态可以配的，所以提供方法供以访问
    @Override
    public int getCoreThreads() {
        if (isActive.get()) {
            if (getRefreshExecutor() != null) {
                return getRefreshExecutor().getCorePoolSize();
            }
        }
        return 0;
    }
	...

ServerListUpdater.UpdateAction：

    public void updateListOfServers() {
        List<T> servers = new ArrayList<T>();
        if (serverListImpl != null) {
            servers = serverListImpl.getUpdatedListOfServers();
            if (filter != null) {
                servers = filter.getFilteredListOfServers(servers);
            }
        }
        updateAllServerList(servers);
    }

@Test
public void fun10() throws InterruptedException {
    ServerListUpdater serverListUpdater = new PollingServerListUpdater();

    serverListUpdater.start(() -> {
        int coreThreads = serverListUpdater.getCoreThreads();
        String lastUpdate = serverListUpdater.getLastUpdate();
        int numberMissedCycles = serverListUpdater.getNumberMissedCycles();
        long durationSinceLastUpdateMs = serverListUpdater.getDurationSinceLastUpdateMs();
        System.out.println("===========上次的执行时间是：" + lastUpdate);
        System.out.println("自上次更新以来已经过的ms数：" + durationSinceLastUpdateMs);
        System.out.println("线程核心数：" + coreThreads);
        System.out.println("错过更新周期的数量：" + numberMissedCycles);

        // .... 执行你对Server列表的更新动作，本处略
    });

    TimeUnit.SECONDS.sleep(500);
}

===========上次的执行时间是：Thu Mar 19 10:28:14 CST 2020
自上次更新以来已经过的ms数：30003
线程核心数：2
错过更新周期的数量：1
===========上次的执行时间是：Thu Mar 19 10:28:44 CST 2020
自上次更新以来已经过的ms数：30002
线程核心数：2
错过更新周期的数量：1
===========上次的执行时间是：Thu Mar 19 10:29:14 CST 2020
自上次更新以来已经过的ms数：30002
线程核心数：2
错过更新周期的数量：1