Nacos疑問之爲何我服務明明下線了卻仍是能夠調用到？

疑問

以前在參與nacos的開發過程當中，有很多同窗都在問，爲何我在nacos console中將服務進行下線了，可是這個被下線的服務仍是能夠被調用到，這不太符合官方宣稱的秒級上下線特色呀。通過進一步詢問發現，那些存在說實例下線後依舊能夠對外提供服務的問題，有一個共同的特色——都有rabbion這個負載均衡的組件。所以本文將從兩個方面探討這個問題：nacos的秒級上下線的實現方式以及rabbion的實例更新機制致使實例上下線感知延遲

Nacos 秒級上下線

@CanDistro
@RequestMapping(value = "", method = RequestMethod.PUT)
public String update(HttpServletRequest request) throws Exception {
	String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
	String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);

	String agent = request.getHeader("Client-Version");
	if (StringUtils.isBlank(agent)) {
		agent = request.getHeader("User-Agent");
	}

	ClientInfo clientInfo = new ClientInfo(agent);

	if (clientInfo.type == ClientInfo.ClientType.JAVA &&
		clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {
		serviceManager.updateInstance(namespaceId, serviceName, parseInstance(request));
	} else {
		serviceManager.registerInstance(namespaceId, serviceName, parseInstance(request));
	}
	return "ok";
}
複製代碼

上面就是nacos console端實例上下線的接口，parseInstance(request)方法就是從request中提取instance實例信息。而背後的updateInstance方法以下

public void updateInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {

	Service service = getService(namespaceId, serviceName);

	if (service == null) {
		throw new NacosException(NacosException.INVALID_PARAM, "service not found, namespace: " + namespaceId + ", service: " + serviceName);
	}

	if (!service.allIPs().contains(instance)) {
		throw new NacosException(NacosException.INVALID_PARAM, "instance not exist: " + instance);
	}

	addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
}

public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips) throws NacosException {

	String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);

	Service service = getService(namespaceId, serviceName);

	List<Instance> instanceList = addIpAddresses(service, ephemeral, ips);

	Instances instances = new Instances();
	instances.setInstanceList(instanceList);

	consistencyService.put(key, instances);
}
複製代碼

接下來的方法就和以前的博文Nacos Server端註冊一個服務實例流程同樣了。所以在nacos console中一旦點擊實例下線，是立馬更新nacos naming server中的實例信息數據的。

Rabbion的實例更新機制

首先看nacos實現的rabbion的實例拉取代碼

public class NacosServerList extends AbstractServerList<NacosServer> {

	private NacosDiscoveryProperties discoveryProperties;

	private String serviceId;

	public NacosServerList(NacosDiscoveryProperties discoveryProperties) {
		this.discoveryProperties = discoveryProperties;
	}

	@Override
	public List<NacosServer> getInitialListOfServers() {
		return getServers();
	}

	@Override
	public List<NacosServer> getUpdatedListOfServers() {
		return getServers();
	}

	private List<NacosServer> getServers() {
		try {
			List<Instance> instances = discoveryProperties.namingServiceInstance()
					.selectInstances(serviceId, true);
			return instancesToServerList(instances);
		}
		catch (Exception e) {
			throw new IllegalStateException(
					"Can not get service instances from nacos, serviceId=" + serviceId,
					e);
		}
	}

	private List<NacosServer> instancesToServerList(List<Instance> instances) {
		List<NacosServer> result = new ArrayList<>();
		if (null == instances) {
			return result;
		}
		for (Instance instance : instances) {
			result.add(new NacosServer(instance));
		}

		return result;
	}

	public String getServiceId() {
		return serviceId;
	}

	@Override
	public void initWithNiwsConfig(IClientConfig iClientConfig) {
		this.serviceId = iClientConfig.getClientName();
	}
}
複製代碼

能夠看到NacosServerList繼承了AbstractServerList，那麼這個AbstractServerList最終在哪裏被收集呢？經過代碼跟蹤能夠看到，最終是在DynamicServerListLoadBalancer這個類中被收集

protected final ServerListUpdater.UpdateAction updateAction = new ServerListUpdater.UpdateAction() {
        @Override
        public void doUpdate() {
            updateListOfServers();
        }
    };

public DynamicServerListLoadBalancer(IClientConfig clientConfig) {
	initWithNiwsConfig(clientConfig);
}
    
@Override
public void initWithNiwsConfig(IClientConfig clientConfig) {
	try {
		super.initWithNiwsConfig(clientConfig);
		String niwsServerListClassName = clientConfig.getPropertyAsString( CommonClientConfigKey.NIWSServerListClassName, DefaultClientConfigImpl.DEFAULT_SEVER_LIST_CLASS);
		ServerList<T> niwsServerListImpl = (ServerList<T>) ClientFactory
                    .instantiateInstanceWithClientConfig(niwsServerListClassName, clientConfig);
    // 獲取全部ServerList接口的實現類
		this.serverListImpl = niwsServerListImpl;

    // 獲取Filter(對拉取的servers列表實行過濾操做)
		if (niwsServerListImpl instanceof AbstractServerList) {
			AbstractServerListFilter<T> niwsFilter = ((AbstractServerList) niwsServerListImpl)
                        .getFilterImpl(clientConfig);
			niwsFilter.setLoadBalancerStats(getLoadBalancerStats());
			this.filter = niwsFilter;
		}

    // 獲取獲取ServerListUpdater對象實現類類名
		String serverListUpdaterClassName = clientConfig.getPropertyAsString( CommonClientConfigKey.ServerListUpdaterClassName, DefaultClientConfigImpl.DEFAULT_SERVER_LIST_UPDATER_CLASS);

    // 獲取ServerListUpdater對象(實際對象爲PollingServerListUpdater)
		this.serverListUpdater = (ServerListUpdater) ClientFactory.instantiateInstanceWithClientConfig(serverListUpdaterClassName, clientConfig);

    // 初始化或者重置
		restOfInit(clientConfig);
	} catch (Exception e) {
		throw new RuntimeException(
                    "Exception while initializing NIWSDiscoveryLoadBalancer:"
                            + clientConfig.getClientName()
                            + ", niwsClientConfig:" + clientConfig, e);
	}
}

void restOfInit(IClientConfig clientConfig) {
	boolean primeConnection = this.isEnablePrimingConnections();
	// turn this off to avoid duplicated asynchronous priming done in BaseLoadBalancer.setServerList()
	this.setEnablePrimingConnections(false);
  // 開啓定時任務，這個任務就是定時刷新實例信息緩存
	enableAndInitLearnNewServersFeature();

  // 開啓前進行一次實例拉取操做
	updateListOfServers();
	if (primeConnection && this.getPrimeConnections() != null) {
		this.getPrimeConnections() .primeConnections(getReachableServers());
	}
	this.setEnablePrimingConnections(primeConnection);
	LOGGER.info("DynamicServerListLoadBalancer for client {} initialized: {}", clientConfig.getClientName(), this.toString());
}

// 這裏就是進行實例信息緩存更新的操做
@VisibleForTesting
public void updateListOfServers() {
	List<T> servers = new ArrayList<T>();
	if (serverListImpl != null) {
    // 調用拉取新實例信息的方法
		servers = serverListImpl.getUpdatedListOfServers();
		LOGGER.debug("List of Servers for {} obtained from Discovery client: {}", getIdentifier(), servers);

    // 用Filter對拉取的servers列表進行更新
		if (filter != null) {
			servers = filter.getFilteredListOfServers(servers);
			LOGGER.debug("Filtered List of Servers for {} obtained from Discovery client: {}", getIdentifier(), servers);
		}
	}
  // 更新實例列表
	updateAllServerList(servers);
}
複製代碼

來看看enableAndInitLearnNewServersFeature();的最終調用是什麼

@Override
public synchronized void start(final UpdateAction updateAction) {
	if (isActive.compareAndSet(false, true)) {
		final Runnable wrapperRunnable = new Runnable() {
			@Override
			public void run() {
				if (!isActive.get()) {
					if (scheduledFuture != null) {
						scheduledFuture.cancel(true);
					}
					return;
				}
				try {
          // 這裏的UpdateAction對象就是在DynamicServerListLoadBalancer中封裝的updateListOfServers實現
					updateAction.doUpdate();
					lastUpdated = System.currentTimeMillis();
				} catch (Exception e) {
					logger.warn("Failed one update cycle", e);
				}
			}
		};

    // 默認任務執行時間間隔爲30s
		scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay(
                    wrapperRunnable,
                    initialDelayMs,
                    refreshIntervalMs,
                    TimeUnit.MILLISECONDS);
	} else {
		logger.info("Already active, no-op");
	}
}
複製代碼

所以不難看出，雖然nacos實現了秒級的實例上下線，可是因爲在Spring Cloud中，負載組件rabbion的實例信息更新是採用了定時任務的形式，有可能這個任務上一秒剛剛執行完，下一秒你就執行實例上下線操做，那麼rabbion要感知這個變化，就必需要等待refreshIntervalMs秒後才能夠感知到。