在分布式服务中,原来的单体服务会被拆分成一个个微服务,服务注册实例到注册中心,服务消费者通过注册中心获取实例列表,直接请求调用服务。
服务是如何注册到注册中心,服务如果挂了,服务是如何检测?带着这些问题,我们从源码上对服务注册进行简单的源码分析。
方便统一版本,客户端和服务端版本号都为2.1.1。
启动nacos服务注册和发现需要添加maven依赖:
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
<version>${latest.version}</version>
</dependency>
根据maven依赖找到对应的spring.factories文件:
在spring.factories文件里找到启动配置类信息,SpringBoot服务启动时会将这些配置类信息注入到bean容器中。
org.springframework.boot.autoconfigure.EnableAutoConfiguration=
com.alibaba.cloud.nacos.NacosDiscoveryAutoConfiguration,
com.alibaba.cloud.nacos.ribbon.RibbonNacosAutoConfiguration,
com.alibaba.cloud.nacos.endpoint.NacosDiscoveryEndpointAutoConfiguration,
com.alibaba.cloud.nacos.discovery.NacosDiscoveryClientAutoConfiguration,
com.alibaba.cloud.nacos.discovery.configclient.NacosConfigServerAutoConfiguration
org.springframework.cloud.bootstrap.BootstrapConfiguration=
com.alibaba.cloud.nacos.discovery.configclient.NacosDiscoveryClientConfigServiceBootstrapConfiguration
服务注册的核心配置类为:NacosDiscoveryAutoConfiguration,该类配置三个bean对象:
NacosAutoServiceRegistration继承了抽象类AbstractAutoServiceRegistration。AbstractAutoServiceRegistration抽象类又实现了ApplicationListener接口。
实现ApplicationListener接口的方法,会在Spring容器初始化完成之后调用onApplicationEvent方法:
public void onApplicationEvent(WebServerInitializedEvent event) {
bind(event);
}
调用bind方法:
public void bind(WebServerInitializedEvent event) {
ApplicationContext context = event.getApplicationContext();
if (context instanceof ConfigurableWebServerApplicationContext) {
if ("management".equals(((ConfigurableWebServerApplicationContext) context)
.getServerNamespace())) {
return;
}
}
this.port.compareAndSet(0, event.getWebServer().getPort());
// 调用 start 方法
this.start();
}
调用了start方法:
public void start() {
if (!isEnabled()) {
if (logger.isDebugEnabled()) {
logger.debug("Discovery Lifecycle disabled. Not starting");
}
return;
}
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);
}
}
调用了register方法,最终调用的是NacosServiceRegistry类的register方法。
根据上文可知,服务器启动后调用NacosServiceRegistry类的register方法,该方法实现将实例注册到服务端:
public void register(Registration registration) {
if (StringUtils.isEmpty(registration.getServiceId())) {
log.warn("No service to register for nacos client...");
return;
}
String serviceId = registration.getServiceId();
String group = nacosDiscoveryProperties.getGroup();
// 创建实例
Instance instance = getNacosInstanceFromRegistration(registration);
try {
// 注册实例
namingService.registerInstance(serviceId, group, instance);
log.info("nacos registry, {} {} {}:{} register finished", group, serviceId,
instance.getIp(), instance.getPort());
}
catch (Exception e) {
log.error("nacos registry, {} register fAIled...{},", serviceId,
registration.toString(), e);
}
}
创建实例,然后通过namingService.registerInstance方法注册实例,然后查看registerInstance方法:
@Override
public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
if (instance.isEphemeral()) {
// 封装心跳包
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);
long instanceInterval = instance.getInstanceHeartBeatInterval();
beatInfo.setPeriod(instanceInterval == 0 ? DEFAULT_HEART_BEAT_INTERVAL : instanceInterval);
// 发送心跳包
beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);
}
// 发送实例
serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance);
}
registerInstance主要做两件事:
beatReactor.addBeatInfo使用定时服务,每隔5s向服务端发送一次心跳请求,通过http请求发送心跳信息,路径为/v1/ns/instance/beat。
心跳请求定时任务使用线程池ScheduledThreadPoolExecutor.schedule(),而该方法只会调用一次,定时任务的实现是在每次请求任务只会再调用一次ScheduledThreadPoolExecutor.schedule(), 简单说就是nacos在发送心跳的时候,会调用schedule方法,在schedule要执行的任务中,如果正常发送完心跳,会再次调用schedule方法。
那为什么不直接调用周期执行的线程池ScheduledThreadPoolExecutor.scheduleAtFixedRate()?可能是由于发送心跳服务发生异常后,定时任务还会继续执行,但是周期执行的线程池遇到报错后也不会重复调用执行的任务。
线程任务BeatTask的run方法,,每次执行会先判断isStopped,如果是false,说明心跳停止,就不会触发下次执行任务。如果使用定时任务scheduleAtFixedRate,即使心跳停止还会继续执行任务,造成资源不必要浪费。
registerService主要封装实例信息,比如ip、port、servicename,将这些信息通过http请求发送给服务端。路径为/v1/ns/instance。
根据上面流程,查看以下的流程图:
服务端就是注册中心,服务注册到注册中心,在https://Github.com/alibaba/nacos/releases/tag/2.1.1下载源码部署到本地,方便调式和查看,部署方式详见我的另外一篇文章Nacos 源码环境搭建。
服务端主要接收两个信息:心跳包和实例信息。
客户端向服务请求的路径为/v1/ns/instance/beat,对应的服务端为InstanceController类的beat方法:
@PutMapping("/beat")
@Secured(action = ActionTypes.WRITE)
public ObjectNode beat(HttpServletRequest request) throws Exception {
ObjectNode result = JacksonUtils.createEmptyJsonNode();
result.put(SwitchEntry.CLIENT_BEAT_INTERVAL, switchDomain.getClientBeatInterval());
String beat = WebUtils.optional(request, "beat", StringUtils.EMPTY);
RsInfo clientBeat = null;
// 判断是否有心跳,存在心跳就转成RsInfo
if (StringUtils.isNotBlank(beat)) {
clientBeat = JacksonUtils.toObj(beat, RsInfo.class);
}
String clusterName = WebUtils
.optional(request, CommonParams.CLUSTER_NAME, UtilsAndCommons.DEFAULT_CLUSTER_NAME);
String ip = WebUtils.optional(request, "ip", StringUtils.EMPTY);
int port = Integer.parseInt(WebUtils.optional(request, "port", "0"));
if (clientBeat != null) {
if (StringUtils.isNotBlank(clientBeat.getCluster())) {
clusterName = clientBeat.getCluster();
} else {
// fix #2533
clientBeat.setCluster(clusterName);
}
ip = clientBeat.getIp();
port = clientBeat.getPort();
}
String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
NamingUtils.checkServiceNameFormat(serviceName);
Loggers.SRV_LOG.debug("[CLIENT-BEAT] full arguments: beat: {}, serviceName: {}, namespaceId: {}", clientBeat,
serviceName, namespaceId);
// 获取实例信息
BeatInfoInstanceBuilder builder = BeatInfoInstanceBuilder.newBuilder();
builder.setRequest(request);
int resultCode = getInstanceOperator()
.handleBeat(namespaceId, serviceName, ip, port, clusterName, clientBeat, builder);
result.put(CommonParams.CODE, resultCode);
// 下次发送心跳包间隔
result.put(SwitchEntry.CLIENT_BEAT_INTERVAL,
getInstanceOperator().getHeartBeatInterval(namespaceId, serviceName, ip, port, clusterName));
result.put(SwitchEntry.LIGHT_BEAT_ENABLED, switchDomain.isLightBeatEnabled());
return result;
}
在handleBeat方法中执行线程任务ClientBeatProcessorV2的run方法,延长lastHeartBeatTime时间。注册中心会定时查询实例,当前时间 - lastHeartBeatTime > 设置时间(默认15秒),就标记实例为不健康实例。如果心跳实例不健康,发送通知给订阅方,变更实例。
服务端在15秒没有收到心跳包会将实例设置为不健康,在30秒没有收到心跳包会将临时实例移除掉。
客户端请求的地址是/nacos/v1/ns/instance, 对应的是服务端是在InstanceController类。找到类上对应的post请求方法上。
注册流程:
InstanceController#register ——>InstanceOperatorClientImpl#registerInstance ——>ClientOperationServiceProxy#registerInstance ——>EphemeralClientOperationServiceImpl#registerInstance
服务注册后,将服务存储在一个双层map集合中:
private final Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>();
通过是否存在ephemeral,为true,走AP模式,否则走CP模式。
Nacos 默认就是采用的AP模式使用Distro协议实现。实现的接口是EphemeralConsistencyService对节点信息的持久化主要是调用put方法,
会先写入到DataStore中:
public void onPut(String key, Record value) {
if (KeyBuilder.matchEphemeralInstanceListKey(key)) {
Datum<Instances> datum = new Datum<>();
datum.value = (Instances) value;
datum.key = key;
datum.timestamp.incrementAndGet();
// 数据持久化到缓存中
dataStore.put(key, datum);
}
if (!listeners.containsKey(key)) {
return;
}
notifier.addTask(key, DataOperation.CHANGE);
}