The principle of CAP
C: consistency
A: Availability (Avaliablelity)
P: Partition fault tolerance
The first thing a distributed system must ensure is that most P’s choose between C and P
Zone fault tolerance: Generally, system zones deployed for multiple borrowing points refer to network zones (contacts cannot communicate and synchronize data due to network reasons). Fault tolerance refers to system nodes that are partitioned must still be able to provide services. It cannot be said that the partition causes the whole system to break down and fail to provide services
Nacos operating mode
Nacos supports AP CP mode
Guarantee AP mode: When a service fails to synchronize data between services due to the breakdown of a service, data consistency is sacrificed to ensure system availability.
Persistent nodes and temporary nodes
\
Nacos registration process
First load it from the Spring.factories file in SpringBoot
NacosServiceRegistryAutoConfiguration
/ * * registry autoServiceRegistrationProperties NacosAutoServiceRegistration by @ Autower * * * / @ Bean @ConditionalOnBean(AutoServiceRegistrationProperties.class) public NacosAutoServiceRegistration nacosAutoServiceRegistration( NacosServiceRegistry registry, AutoServiceRegistrationProperties autoServiceRegistrationProperties, NacosRegistration registration) { return new NacosAutoServiceRegistration(registry, autoServiceRegistrationProperties, registration); } } //return new NacosAutoServiceRegistration(registry, // autoServiceRegistrationProperties, registration); public NacosAutoServiceRegistration(ServiceRegistry<Registration> serviceRegistry, AutoServiceRegistrationProperties autoServiceRegistrationProperties, NacosRegistration registration) {/ / the main is to call the superclass / / AbstractAutoServiceRegistration < registration > the constructor of the incoming serviceRegistry super(serviceRegistry, autoServiceRegistrationProperties); this.registration = registration; } / / while listeners AbstractAutoServiceRegistration ApplicationListener / / serviceRegistry is NacosServiceRegistry instance is in the initialization phase (currently only guess) protected AbstractAutoServiceRegistration(ServiceRegistry<R> serviceRegistry, AutoServiceRegistrationProperties properties) { this.serviceRegistry = serviceRegistry; this.properties = properties; }Copy the code//public abstract class AbstractAutoServiceRegistration<R extends Registration> //implements AutoServiceRegistration, ApplicationContextAware, ApplicationListener < WebServerInitializedEvent > implement the listener interface to realize the @ Override @ SuppressWarnings (" deprecation ") public void onApplicationEvent(WebServerInitializedEvent event) { bind(event); } @Deprecated 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()); // Call the start method this.start(); } public void start() { if (! isEnabled()) { if (logger.isDebugEnabled()) { logger.debug("Discovery Lifecycle disabled. Not starting"); } return; } // only initialize if nonSecurePort is greater than 0 and it isn't already running // because of containerPortInitializer below if (! This. Running. The get ()) {/ / publish event enclosing context. PublishEvent (new InstancePreRegisteredEvent (this, getRegistration ())); register(); if (shouldRegisterManagement()) { registerManagement(); } / / publish event enclosing context. PublishEvent (new InstanceRegisteredEvent < > (this, getConfiguration ())); this.running.compareAndSet(false, true); }}Copy the codecom.alibaba.cloud.nacos.registry.NacosServiceRegistry
Heartbeat Registration information
The heartbeat is sent and the service is registered at the same time
Just updated a field lastBeat
@Override
public void register(Registration registration) {
if (StringUtils.isEmpty(registration.getServiceId())) {
log.warn("No service to register for nacos client...");
return;
}
NamingService namingService = namingService();
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);
// rethrow a RuntimeException if the registration is failed.
// issue : https://github.com/alibaba/spring-cloud-alibaba/issues/1132
rethrowRuntimeException(e);
}
}
//namingService.registerInstance(serviceId, group, instance);
@Override
public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
NamingUtils.checkInstanceIsLegal(instance);
String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
if (instance.isEphemeral()) {
//构建Beat心跳信息
BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);
beatReactor.addBeatInfo(groupedServiceName, beatInfo);
}
serverProxy.registerService(groupedServiceName, groupName, instance);
}
// beatReactor.addBeatInfo(groupedServiceName, beatInfo);
public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo);
String key = buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort());
BeatInfo existBeat = null;
//fix #1733
if ((existBeat = dom2Beat.remove(key)) != null) {
existBeat.setStopped(true);
}
dom2Beat.put(key, beatInfo);
//通过线程池的方式轮询去发送心跳
executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
MetricsMonitor.getDom2BeatSizeMonitor().set(dom2Beat.size());
}
@Override
public void run() {
if (beatInfo.isStopped()) {
return;
}
long nextTime = beatInfo.getPeriod();
try {
//发送心跳信息
JsonNode result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled);
long interval = result.get("clientBeatInterval").asLong();
boolean lightBeatEnabled = false;
if (result.has(CommonParams.LIGHT_BEAT_ENABLED)) {
lightBeatEnabled = result.get(CommonParams.LIGHT_BEAT_ENABLED).asBoolean();
}
BeatReactor.this.lightBeatEnabled = lightBeatEnabled;
if (interval > 0) {
nextTime = interval;
}
int code = NamingResponseCode.OK;
if (result.has(CommonParams.CODE)) {
code = result.get(CommonParams.CODE).asInt();
}
if (code == NamingResponseCode.RESOURCE_NOT_FOUND) {
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 {
//注册服务 通过Http请求向Nacos发送请求，进行服务的注册
serverProxy.registerService(beatInfo.getServiceName(),
NamingUtils.getGroupName(beatInfo.getServiceName()), instance);
} catch (Exception ignore) {
}
}
} catch (NacosException ex) {
NAMING_LOGGER.error("[CLIENT-BEAT] failed to send beat: {}, code: {}, msg: {}",
JacksonUtils.toJson(beatInfo), ex.getErrCode(), ex.getErrMsg());
}
//再次发送心跳 会过每五秒执行一次
executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS);
}
//serverProxy.registerService 心跳注册
/****
nacos/v1/ns/instance
***/
/**
* Send beat.
*
* @param beatInfo beat info
* @param lightBeatEnabled light beat
* @return beat result
* @throws NacosException nacos exception
*/
public JsonNode sendBeat(BeatInfo beatInfo, boolean lightBeatEnabled) throws NacosException {
if (NAMING_LOGGER.isDebugEnabled()) {
NAMING_LOGGER.debug("[BEAT] {} sending beat to server: {}", namespaceId, beatInfo.toString());
}
Map<String, String> params = new HashMap<String, String>(8);
Map<String, String> bodyMap = new HashMap<String, String>(2);
if (!lightBeatEnabled) {
bodyMap.put("beat", JacksonUtils.toJson(beatInfo));
}
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()));
String result = reqApi(UtilAndComs.nacosUrlBase + "/instance/beat", params, bodyMap, HttpMethod.PUT);
return JacksonUtils.toObj(result);
}
/** String result = reqApi(UtilAndComs.nacosUrlBase + "/instance/beat", params, bodyMap, HttpMethod.PUT);
* Request api.
*
* @param api api
* @param params parameters
* @param body body
* @param servers servers
* @param method http method
* @return result
* @throws NacosException nacos exception
*/
public String reqApi(String api, Map<String, String> params, Map<String, String> body, List<String> servers,
String method) throws NacosException {
params.put(CommonParams.NAMESPACE_ID, getNamespaceId());
if (CollectionUtils.isEmpty(servers) && StringUtils.isBlank(nacosDomain)) {
throw new NacosException(NacosException.INVALID_PARAM, "no server available");
}
NacosException exception = new NacosException();
if (StringUtils.isNotBlank(nacosDomain)) {
for (int i = 0; i < maxRetry; i++) {
try {
return callServer(api, params, body, nacosDomain, method);
} catch (NacosException e) {
exception = e;
if (NAMING_LOGGER.isDebugEnabled()) {
NAMING_LOGGER.debug("request {} failed.", nacosDomain, e);
}
}
}
} else {
//通过随机方式向服务进行注册，主要目的是为了将请求随机分配给不同的Nacos服务
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();
}
}
NAMING_LOGGER.error("request: {} failed, servers: {}, code: {}, msg: {}", api, servers, exception.getErrCode(),
exception.getErrMsg());
throw new NacosException(exception.getErrCode(),
"failed to req API:" + api + " after all servers(" + servers + ") tried: " + exception.getMessage());
}
Copy the codeThe service registry
Go to the server to find the registration
- Temporary nodes support the Ephemeral field
- The default is to register temporary nodes
**
*注册服务实例接口
Register new instance.
*
* @param request http request
* @return 'ok' if success
* @throws Exception any error during register
*/
@CanDistro
@PostMapping
@Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
public String register(HttpServletRequest request) throws Exception {
final String namespaceId = WebUtils
.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
NamingUtils.checkServiceNameFormat(serviceName);
final Instance instance = parseInstance(request);
getInstanceOperator().registerInstance(namespaceId, serviceName, instance);
return "ok";
}
/**注册的核心方法
* Register an instance to a service in AP mode.
*
* <p>This method creates service or cluster silently if they don't exist.
*
* @param namespaceId id of namespace
* @param serviceName service name
* @param instance instance to register
* @throws Exception any error occurred in the process
*/
public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
//从注册Map获取Map信息如果有的话 就从map中获取,,，没有的话就进行注册
createEmptyService(namespaceId, serviceName, instance.isEphemeral());
Service service = getService(namespaceId, serviceName);
if (service == null) {
throw new NacosException(NacosException.INVALID_PARAM,
"service not found, namespace: " + namespaceId + ", service: " + serviceName);
}
//添加实例
addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
}
/**
//createEmptyService(namespaceId, serviceName, instance.isEphemeral());
* Create service if not exist.
*
* @param namespaceId namespace
* @param serviceName service name
* @param local whether create service by local
* @param cluster cluster
* @throws NacosException nacos exception
*/
public void createServiceIfAbsent(String namespaceId, String serviceName, boolean local, Cluster cluster)
throws NacosException {
Service service = getService(namespaceId, serviceName);
if (service == null) {
Loggers.SRV_LOG.info("creating empty service {}:{}", namespaceId, serviceName);
service = new Service();
service.setName(serviceName);
service.setNamespaceId(namespaceId);
service.setGroupName(NamingUtils.getGroupName(serviceName));
// now validate the service. if failed, exception will be thrown
service.setLastModifiedMillis(System.currentTimeMillis());
service.recalculateChecksum();
if (cluster != null) {
cluster.setService(service);
service.getClusterMap().put(cluster.getName(), cluster);
}
service.validate();
putServiceAndInit(service);
if (!local) {
addOrReplaceService(service);
}
}
}
/**
* Add instance to service.
*
* @param namespaceId namespace
* @param serviceName service name
* @param ephemeral whether instance is ephemeral
* @param ips instances
* @throws NacosException nacos exception
*/
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);
synchronized (service) {
List<Instance> instanceList = addIpAddresses(service, ephemeral, ips);
Instances instances = new Instances();
instances.setInstanceList(instanceList);
//Key 定义的一个名字
//value:实例的列表
//把KEY添加到任务队列
//队列操作 只是一个一个的去做
//KeyPoint
consistencyService.put(key, instances);
}
}
// consistencyService.put(key, instances);
@Override
public void put(String key, Record value) throws NacosException {
mapConsistencyService(key).put(key, value);
}
// mapConsistencyService(key).put(key, value);
//如果是临时实例会执行 DistroConsistencyServiceImpl的put方法
//如果是持久实例会执行 RaftConsistencyServiceImpl的put方法
//目前只针对临时实例 DistroConsistencyServiceImpl的put方法
//判断是否是临时节点或者是持久节点 来创建持久节点KEY或者是临时节点KEY
public static String buildInstanceListKey(String namespaceId, String serviceName, boolean ephemeral) {
return ephemeral ? buildEphemeralInstanceListKey(namespaceId, serviceName)
: buildPersistentInstanceListKey(namespaceId, serviceName);
}
//AP结构ephemeralConsistencyService CP架构 persistentConsistencyService
private ConsistencyService mapConsistencyService(String key) {
return KeyBuilder.matchEphemeralKey(key) ? ephemeralConsistencyService : persistentConsistencyService;
}
//目前只针对临时实例 DistroConsistencyServiceImpl的put方法
@Override
public void put(String key, Record value) throws NacosException {
onPut(key, value);
// If upgrade to 2.0.X, do not sync for v1.
if (ApplicationUtils.getBean(UpgradeJudgement.class).isUseGrpcFeatures()) {
return;
}
distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,
DistroConfig.getInstance().getSyncDelayMillis());
}
/**
* onPut(key, value);
*
* @param key key of record
* @param value record
*/
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;
}
//把KEY添加到任务队列
// private volatile Notifier notifier = new Notifier();
// notifier是一个线程
/*
* 这个地方其实是异步操作
* 注册服务
*一边通过异步复制作为副本，然后在把副本合并到真实的注册表map当中
这个里面的每一个副本都是放在队列中的，所以不会存在合并副本造成的多线程的问题
*
* */
notifier.addTask(key, DataOperation.CHANGE);
}
/** notifier.addTask(key, DataOperation.CHANGE);
* Add new notify task to queue.
*
* @param datumKey data key
* @param action action for data
*/
public void addTask(String datumKey, DataOperation action) {
if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {
return;
}
if (action == DataOperation.CHANGE) {
services.put(datumKey, StringUtils.EMPTY);
}
tasks.offer(Pair.with(datumKey, action));
}
Copy the codeKey health check of service
- When the service is initiated, it passes through a thread pool of scheduled tasks
/** * Schedule client beat check task with a delay. * * @param task client beat check task */ public static void scheduleCheck(BeatCheckTask task) { Runnable wrapperTask = task instanceof NacosHealthCheckTask ? new HealthCheckTaskInterceptWrapper((NacosHealthCheckTask) task) : task; futureMap.computeIfAbsent(task.taskKey(), k -> GlobalExecutor.scheduleNamingHealth(task, 5000, 5000, TimeUnit.MILLISECONDS)); } BeatCheckTask Task thread @override public void run() {try {// If upgrade to 2.0.x stop health check with v1 If (ApplicationUtils.getBean(UpgradeJudgement.class).isUseGrpcFeatures()) { return; } if (! getDistroMapper().responsible(service.getName())) { return; } if (! getSwitchDomain().isHealthCheckEnabled()) { return; } List<Instance> instances = service.allIPs(true); // First set health status of instances: for (Instance Instance: instances) { if (System.currentTimeMillis() - instance.getLastBeat() > instance.getInstanceHeartBeatTimeOut()) { if (! Instance.ismarked ()) {if (instance.ishealthy ()) {// Set the health status instance.sethealthy (false); Loggers.EVT_LOG .info("{POS} {IP-DISABLED} valid: {}:{}@{}@{}, region: {}, msg: client timeout after {}, last beat: {}", instance.getIp(), instance.getPort(), instance.getClusterName(), service.getName(), UtilsAndCommons.LOCALHOST_SITE, instance.getInstanceHeartBeatTimeOut(), instance.getLastBeat()); getPushService().serviceChanged(service); } } } } if (! getGlobalConfig().isExpireInstance()) { return; } // then remove obsolete instances: for (Instance instance : instances) { if (instance.isMarked()) { continue; If (system.currentTimemillis () -instance.getLastBeat () > instance.getIpDeleteTimeout()) { // delete instance Loggers.SRV_LOG.info("[AUTO-DELETE-IP] service: {}, ip: {}", service.getName(), JacksonUtils.toJson(instance)); deleteIp(instance); } } } catch (Exception e) { Loggers.SRV_LOG.warn("Exception while processing client beat time out.", e); Private void deleteIP(Instance Instance) {try {namingproxy. Request Request = NamingProxy.Request.newRequest(); request.appendParam("ip", instance.getIp()) .appendParam("port", String.valueOf(instance.getPort())) .appendParam("ephemeral", "true") .appendParam("clusterName", instance.getClusterName()) .appendParam("serviceName", service.getName()) .appendParam("namespaceId", service.getNamespaceId()); The String url = "http://127.0.0.1:" + RunningConfig. GetServerPort () + RunningConfig. GetContextPath () + UtilsAndCommons.NACOS_NAMING_CONTEXT + "/instance?" + request.toUrl(); // delete instance asynchronously: HttpClient.asyncHttpDelete(url, null, null, new AsyncCompletionHandler() { @Override public Object onCompleted(Response response) throws Exception { if (response.getStatusCode() ! = HttpURLConnection.HTTP_OK) { Loggers.SRV_LOG.error("[IP-DEAD] failed to delete ip automatically, ip: {}, caused {}, resp code: {}", instance.toJSON(), response.getResponseBody(), response.getStatusCode()); } return null; }}); } catch (Exception e) { Loggers.SRV_LOG.error("[IP-DEAD] failed to delete ip automatically, ip: {}, error: {}", instance.toJSON(), e); }}Copy the codeService discovery
Question:
The difference between temporary and persistent nodes at service discovery time
@override public List<Instance> getAllInstances(String serviceName, String groupName, List<String> Clusters, boolean subscribe) throws NacosException { ServiceInfo serviceInfo; String clusterString = StringUtils.join(clusters, ","); If (subscribe) {/ / this is actually the client service information registry serviceInfo. = serviceInfoHolder getServiceInfo (serviceName, groupName. clusterString); ServiceInfo = ClientProxy. subscribe(serviceName, groupName, subscribe) {serviceInfo = clientProxy.subscribe(serviceName, groupName, subscribe); clusterString); } } else { serviceInfo = clientProxy.queryInstancesOfService(serviceName, groupName, clusterString, 0, false); } List<Instance> list; if (serviceInfo == null || CollectionUtils.isEmpty(list = serviceInfo.getHosts())) { return new ArrayList<Instance>(); } return list; } / / com. Alibaba. Nacos. Client. Naming. Remote. HTTP. NamingHttpClientProxy / / to the server query service instance @ Override public ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException { return queryInstancesOfService(serviceName, groupName, clusters, pushReceiver.getUdpPort(), false); } /* gwlong * Gets the list of services *.X uses GrpcHttp * * */ @override public ServiceInfo queryInstancesOfService(String serviceName, String groupName, String clusters, int udpPort, boolean healthyOnly) throws NacosException { final Map<String, String> params = new HashMap<String, String>(8); params.put(CommonParams.NAMESPACE_ID, namespaceId); params.put(CommonParams.SERVICE_NAME, NamingUtils.getGroupedName(serviceName, groupName)); params.put("clusters", clusters); params.put("udpPort", String.valueOf(udpPort)); params.put("clientIP", NetUtils.localIP()); params.put("healthyOnly", String.valueOf(healthyOnly)); String result = reqApi(utilandcoms.nacosurlbase + "/instance/list", params, httpmethod.get); if (StringUtils.isNotEmpty(result)) { return JacksonUtils.toObj(result, ServiceInfo.class); } return new ServiceInfo(NamingUtils.getGroupedName(serviceName, groupName), clusters); } # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # server # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # / / get all the service instance @GetMapping("/list") @Secured(parser = NamingResourceParser.class, action = ActionTypes.READ) public Object list(HttpServletRequest request) throws Exception { String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID); String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME); NamingUtils.checkServiceNameFormat(serviceName); String agent = WebUtils.getUserAgent(request); String clusters = WebUtils.optional(request, "clusters", StringUtils.EMPTY); String clientIP = WebUtils.optional(request, "clientIP", StringUtils.EMPTY); int udpPort = Integer.parseInt(WebUtils.optional(request, "udpPort", "0")); boolean healthyOnly = Boolean.parseBoolean(WebUtils.optional(request, "healthyOnly", "false")); boolean isCheck = Boolean.parseBoolean(WebUtils.optional(request, "isCheck", "false")); String app = WebUtils.optional(request, "app", StringUtils.EMPTY); String env = WebUtils.optional(request, "env", StringUtils.EMPTY); String tenant = WebUtils.optional(request, "tid", StringUtils.EMPTY); Subscriber subscriber = new Subscriber(clientIP + ":" + udpPort, agent, app, clientIP, namespaceId, serviceName, udpPort, clusters); return getInstanceOperator().listInstance(namespaceId, serviceName, subscriber, clusters, healthyOnly); }Copy the code\