Nacos简析：distro协议

Posted 2021-03-28 钢笔和椅子

一. CAP理论

• 一致性

• 可用性

• 分区容错性

为什么只能三选二及在分布式场景下P一定不能被放弃？

https://blog.csdn.net/qq_28107929/article/details/106104277

  CAP文氏图 

二. CP的劣势，AP的优势

没有纯粹的CP和AP，保证了CP再尽量靠近A，保证了AP再尽量靠近C。

以RAFT为代表的“强一致性”协议（选主，复制）带来了什么劣势

保障一致性需要做的事情

• 强一致性--->写请求来时日志复制-->主从区分-->选主服务不可用-->投票选主需要一定数量的节点-->可靠性偏低

• 强一致性-->读请求来时请求转发-->主节点压力山大-->性能受限

如果抛弃强一致性我们可以获得什么

• 或许无需选主

• 主节点读写压力分摊

• 不强制要求节点数量即可提供服务

对于服务中心来说，可用，可靠，高性能是第一，反之如果是金融，电商场景，数据一致性则更重要！

三. Distro简介

实现一个AP协议可以参考的对象很少，在nacos-distro出来之前，AP协议代表有以下

• eureka实现的AP协议（去中心化）

• redis-cluster高可用模式（主从复制是异步的）

nacos作为AP派注册中心的粉丝，很多实现上都借鉴了eureka，而Distro作为nacos-ap协议（同时也通过jraft实现raft用作cp，配置中心）的实现，当前仅仅也体现在nacos中开源出来，并没有单独剥离，因此只能说是注册中心的AP协议而非纯粹AP的框架，和etcd的etcd-raft定位很大不同。

distro在nacos中位置

四.Distro部分机制走读

distro部分组件

1. 为什么需要转发，转发到那里，如何转发？

• 每个nacos-server类似分片的数据库，只负责部分实例信息的维护。

• 每个nacos-server类似分片的数据库，只负责部分实例信息的维护。

 private int distroHash(String responsibleTag) { return Math.abs(responsibleTag.hashCode() % Integer.MAX_VALUE); }  

• @CanDistro拦截，调用distroMapper的mapsrv决定转发至哪台nasoc-server，用http形式去转发获取结果。

  简单的distromapper

2.数据初始化，增量同步的基本流程

•  private void load() throws Exception { while (memberManager.allMembersWithoutSelf().isEmpty()) { Loggers.DISTRO.info("[DISTRO-INIT] waiting server list init..."); TimeUnit.SECONDS.sleep(1); } while (distroComponentHolder.getDataStorageTypes().isEmpty()) { Loggers.DISTRO.info("[DISTRO-INIT] waiting distro data storage register..."); TimeUnit.SECONDS.sleep(1); } for (String each : distroComponentHolder.getDataStorageTypes()) { if (!loadCompletedMap.containsKey(each) || !loadCompletedMap.get(each)) { loadCompletedMap.put(each, loadAllDataSnapshotFromRemote(each)); } } }  private boolean loadAllDataSnapshotFromRemote(String resourceType) { DistroTransportAgent transportAgent = distroComponentHolder.findTransportAgent(resourceType); DistroDataProcessor dataProcessor = distroComponentHolder.findDataProcessor(resourceType); if (null == transportAgent || null == dataProcessor) { Loggers.DISTRO.warn("[DISTRO-INIT] Can't find component for type {}, transportAgent: {}, dataProcessor: {}", resourceType, transportAgent, dataProcessor); return false; } for (Member each : memberManager.allMembersWithoutSelf()) { try { Loggers.DISTRO.info("[DISTRO-INIT] load snapshot {} from {}", resourceType, each.getAddress()); DistroData distroData = transportAgent.getDatumSnapshot(each.getAddress()); boolean result = dataProcessor.processSnapshot(distroData); Loggers.DISTRO .info("[DISTRO-INIT] load snapshot {} from {} result: {}", resourceType, each.getAddress(), result); if (result) { return true; } } catch (Exception e) { Loggers.DISTRO.error("[DISTRO-INIT] load snapshot {} from {} failed.", resourceType, each.getAddress(), e); } } return false; }  

• 增量同步通常在dataStore被PUT更新之后（注册实例的最后一步就会调用PUT）再主动触发，因为是异步复制，因此这里只是生成一个延迟的任务，塞入DistroTaskEngine的队列中就马上返回

  
  

   public void put(String key, Record value) throws NacosException { onPut(key, value); //... distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE, globalConfig.getTaskDispatchPeriod() / 2); } public void sync(DistroKey distroKey, DataOperation action, long delay) { for (Member each : memberManager.allMembersWithoutSelf()) { DistroKey distroKeyWithTarget = new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(), each.getAddress()); DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay); distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget, distroDelayTask); //... } } } 

• dataStore的结构长什么样子，同步的数据又长什么样，从全量的快照数据到内存数据中经历了那些操作？

dataStore基于concurrentHashMap存放，其结构较为简单为String--->Datum（Key+Record泛型）

同步的数据开始为[]byte，在Serializer中找到实现Record的class，如instance，通过jackson进行数据转化成期望对象，当前默认nacos-server之间同步的全为instance的数据，获取到数据之后，Distro中需要触发对应listeners。

如instance，在dataStore中存放的key为Instance_LIST_PREFIX+namespaceid+serviceName，value为[]instance