SpringCloud Nacos服务注册和心跳发送(源码)

jar包目录下meta-INF目录下spring.factories文件中

org.springframework.boot.autoconfigure.EnableAutoConfiguration=
  com.alibaba.cloud.nacos.discovery.NacosDiscoveryAutoConfiguration,
  com.alibaba.cloud.nacos.ribbon.RibbonNacosAutoConfiguration,
  com.alibaba.cloud.nacos.endpoint.NacosDiscoveryEndpointAutoConfiguration,
  com.alibaba.cloud.nacos.registry.NacosServiceRegistryAutoConfiguration,
  com.alibaba.cloud.nacos.discovery.NacosDiscoveryClientConfiguration,
  com.alibaba.cloud.nacos.discovery.reactive.NacosReactiveDiscoveryClientConfiguration,
  com.alibaba.cloud.nacos.discovery.configclient.NacosConfigServerAutoConfiguration
org.springframework.cloud.bootstrap.BootstrapConfiguration=
  com.alibaba.cloud.nacos.discovery.configclient.NacosDiscoveryClientConfigServiceBootstrapConfiguration

自动装配com.alibaba.cloud.nacos.registry.NacosServiceRegistryAutoConfiguration

@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties
// 默认注解开启服务发现 @ConditionalOnProperty(value = "spring.cloud.nacos.discovery.enabled", matchIfMissing = true)
@ConditionalOnNacosDiscoveryEnabled
@ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true)
// 在各个类加载后加载当前类
@AutoConfigureAfter({ AutoServiceRegistrationConfiguration.class,
		AutoServiceRegistrationAutoConfiguration.class,
		NacosDiscoveryAutoConfiguration.class })
public class NacosServiceRegistryAutoConfiguration {

	...

	@Bean
	@ConditionalOnBean(AutoServiceRegistrationProperties.class)
	public NacosAutoServiceRegistration nacosAutoServiceRegistration(
			NacosServiceRegistry registry,
			AutoServiceRegistrationProperties autoServiceRegistrationProperties,
			NacosRegistration registration) {
		return new NacosAutoServiceRegistration(registry,
				autoServiceRegistrationProperties, registration);
	}

}

继承了抽象类 AbstractAutoServiceRegistration, 而抽象类又实现了ApplicationListener接口, 当spring refresh完成时会调用对应的onApplicationEvent方法, 开始注册

@Override
@SuppressWarnings("deprecation")
public void onApplicationEvent(WebServerInitializedEvent event) {
	// 服务注册入口
	bind(event);
}

@Deprecated
public void bind(WebServerInitializedEvent event) {
	...
	// 开始注册
	this.start();
}

public void start() {
	...

	if (!this.running.get()) {
		// 注册之前事件发布
		this.context.publishEvent(
				new InstancePreRegisteredEvent(this, getRegistration()));
		// 注册
		register();
		if (shouldRegisterManagement()) {
			registerManagement();
		}
		// 注册之后事件发布
		this.context.publishEvent(
				new InstanceRegisteredEvent<>(this, getConfiguration()));
		// 标记状态
		this.running.compareAndSet(false, true);
	}

}

protected void register() {
	this.serviceRegistry.register(getRegistration());
}

调用NacosServiceRegistry的register方法开始执行注册, 实现了服务注册的接口, 包含服务注册/下线/关闭/状态变更及获取等方法

@Override
public void register(Registration registration) {
	...

	// 设置一系列参数
	String serviceId = registration.getServiceId();
	String group = nacosDiscoveryProperties.getGroup();
	// ephemeral 默认为true 下一个方法会校验
	Instance instance = getNacosInstanceFromRegistration(registration);
	namingService.registerInstance(serviceId, group, instance);
	...
}

NacosNamingService
先注册心跳定时任务, 默认5s后执行, 紧接着去注册服务. 虽然程序上是先发心跳再注册, 实际是先注册再发送心跳.

@Override
public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
	// 判断服务是否为临时 默认为true
    if (instance.isEphemeral()) {
    	// 包含心跳定时为5s
        BeatInfo beatInfo = new BeatInfo();
        beatInfo.setServiceName(NamingUtils.getGroupedName(serviceName, groupName));
        beatInfo.setIp(instance.getIp());
        beatInfo.setPort(instance.getPort());
        beatInfo.setCluster(instance.getClusterName());
        beatInfo.setWeight(instance.getWeight());
        beatInfo.setmetadata(instance.getmetadata());
        beatInfo.setScheduled(false);
        beatInfo.setPeriod(instance.getInstanceHeartBeatInterval());
        // 发起心跳
        beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);
    }
    // 注册服务
    serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance);
}

使用一个带定时的执行器来定时发送心跳, 循环发送并创建定时任务

public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
    ...
    // 添加一个定时任务 默认间隔是5s
    executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
    ...
}

class BeatTask implements Runnable {

    BeatInfo beatInfo;

    public BeatTask(BeatInfo beatInfo) {
        this.beatInfo = beatInfo;
    }

    @Override
    public void run() {
        if (beatInfo.isStopped()) {
            return;
        }
        // 默认间隔
        long nextTime = beatInfo.getPeriod();
        try {
        	// 发送心跳
            JSONObject result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled);
            // 根据结果做相应处理
            long interval = result.getIntValue("clientBeatInterval");
            boolean lightBeatEnabled = false;
            if (result.containsKey(CommonParams.LIGHT_BEAT_ENABLED)) {
                lightBeatEnabled = result.getBooleanValue(CommonParams.LIGHT_BEAT_ENABLED);
            }
            BeatReactor.this.lightBeatEnabled = lightBeatEnabled;
            if (interval > 0) {
                nextTime = interval;
            }
            int code = NamingResponseCode.OK;
            if (result.containsKey(CommonParams.CODE)) {
                code = result.getIntValue(CommonParams.CODE);
            }
            if (code == NamingResponseCode.RESOURCE_NOT_FOUND) {
            	// 如果返回状态码为20404 表示服务已经下线
                Instance instance = new Instance();
                instance.setPort(beatInfo.getPort());
                instance.setIp(beatInfo.getIp());
                instance.setWeight(beatInfo.getWeight());
                instance.setmetadata(beatInfo.getmetadata());
                instance.setClusterName(beatInfo.getCluster());
                instance.setServiceName(beatInfo.getServiceName());
                instance.setInstanceId(instance.getInstanceId());
                instance.setEphemeral(true);
                try {
                	// 重新发起注册
                    serverProxy.registerService(beatInfo.getServiceName(),
                        NamingUtils.getGroupName(beatInfo.getServiceName()), instance);
                } catch (Exception ignore) {
                }
            }
        } catch (NacosException ne) {
            NAMING_LOGGER.error("[CLIENT-BEAT] failed to send beat: {}, code: {}, msg: {}",
                JSON.toJSONString(beatInfo), ne.getErrCode(), ne.getErrMsg());

        }
        // 注册下一个心跳任务, 循环发送
        executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS);
    }
}

public JSONObject sendBeat(BeatInfo beatInfo, boolean lightBeatEnabled) throws NacosException {
	// 构造请求
    if (NAMING_LOGGER.isDebugEnabled()) {
        NAMING_LOGGER.debug("[BEAT] {} sending beat to server: {}", namespaceId, beatInfo.toString());
    }
    Map params = new HashMap(8);
    String body = StringUtils.EMPTY;
    if (!lightBeatEnabled) {
        try {
            body = "beat=" + URLEncoder.encode(JSON.toJSONString(beatInfo), "UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new NacosException(NacosException.SERVER_ERROR, "encode beatInfo error", e);
        }
    }
    params.put(CommonParams.NAMESPACE_ID, namespaceId);
    params.put(CommonParams.SERVICE_NAME, beatInfo.getServiceName());
    params.put(CommonParams.CLUSTER_NAME, beatInfo.getCluster());
    params.put("ip", beatInfo.getIp());
    params.put("port", String.valueOf(beatInfo.getPort()));
    // reqAPI 是通用方法 服务注册也是调用此方法 后面提供源码
    String result = reqAPI(UtilAndComs.NACOS_URL_base + "/instance/beat", params, body, HttpMethod.PUT);
    return JSON.parseObject(result);
}

public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
	// 构造请求
    final Map params = new HashMap(9);
    params.put(CommonParams.NAMESPACE_ID, namespaceId);
    params.put(CommonParams.SERVICE_NAME, serviceName);
    params.put(CommonParams.GROUP_NAME, groupName);
    params.put(CommonParams.CLUSTER_NAME, instance.getClusterName());
    params.put("ip", instance.getIp());
    params.put("port", String.valueOf(instance.getPort()));
    params.put("weight", String.valueOf(instance.getWeight()));
    params.put("enable", String.valueOf(instance.isEnabled()));
    params.put("healthy", String.valueOf(instance.isHealthy()));
    params.put("ephemeral", String.valueOf(instance.isEphemeral()));
    params.put("metadata", JSON.toJSONString(instance.getmetadata()));
    // 发送请求
    reqAPI(UtilAndComs.NACOS_URL_INSTANCE, params, HttpMethod.POST);
}

通用reqAPI方法, 提供http请求, 其中参数servers为nacos服务地址列表

public String reqAPI(String api, Map params, String body, List servers, String method) throws NacosException {
    params.put(CommonParams.NAMESPACE_ID, getNamespaceId());
    if (CollectionUtils.isEmpty(servers) && StringUtils.isEmpty(nacosDomain)) {
        throw new NacosException(NacosException.INVALID_PARAM, "no server available");
    }

    NacosException exception = new NacosException();
    if (servers != null && !servers.isEmpty()) {

        Random random = new Random(System.currentTimeMillis());
        // 随机找一个发送请求
        int index = random.nextInt(servers.size());

        // 从第一个随机开始, 逐个循环请求, 如果有成功就返回, 否则循环请求
        for (int i = 0; i < servers.size(); i++) {
            String server = servers.get(index);
            try {
                return callServer(api, params, body, server, method);
            } catch (NacosException e) {
                exception = e;
                if (NAMING_LOGGER.isDebugEnabled()) {
                    NAMING_LOGGER.debug("request {} failed.", server, e);
                }
            }
            index = (index + 1) % servers.size();
        }
    }
    ...
}

public String callServer(String api, Map params, String body, String curServer, String method)
    throws NacosException {
    // 计时
    long start = System.currentTimeMillis();
    long end = 0;
    // 安全方面 加密
    injectSecurityInfo(params);
    // 请求头
    List headers = builderHeaders();

    // 拼装url 根据配置参数生成完成的url
    String url;
    if (curServer.startsWith(UtilAndComs.HTTPS) || curServer.startsWith(UtilAndComs.HTTP)) {
        url = curServer + api;
    } else {
        if (!curServer.contains(UtilAndComs.SERVER_ADDR_IP_SPLITER)) {
            curServer = curServer + UtilAndComs.SERVER_ADDR_IP_SPLITER + serverPort;
        }
        url = HttpClient.getPrefix() + curServer + api;
    }

    // 请求server
    HttpClient.HttpResult result = HttpClient.request(url, headers, params, body, UtilAndComs.ENCODING, method);
    end = System.currentTimeMillis();

    MetricsMonitor.getNamingRequestMonitor(method, url, String.valueOf(result.code))
        .observe(end - start);

    if (HttpURLConnection.HTTP_OK == result.code) {
        return result.content;
    }

    if (HttpURLConnection.HTTP_NOT_MODIFIED == result.code) {
        return StringUtils.EMPTY;
    }

    throw new NacosException(result.code, result.content);
}

至此, 服务已经完成注册, 并循环发起心跳