2020年11月10日星期二

nacos服务注册之服务器端Distro

一致性协议算法Distro阿里自己的创的算法吧，网上能找到的资料很少。Distro用于处理ephemeral类型数据

  1. Distro协议算法看代码大体流程是：
  • nacos启动首先从其他远程节点同步全部数据
  • nacos每个节点是平等的都可以处理写入请求，同时把新数据同步到其他节点
  • 每个节点只负责部分数据，定时发送自己负责数据校验值到其他节点来保持数据一致性

  1. Distro代码分析， nacos负责数据一致性的服务是ConsistencyService接口，DistroConsistencyServiceImpl是EphemeralConsistencyService和ConsistencyService实现类。

    Distro协议算法数据存储实现笔记简单，DataStore是存储实现类临时节点的数据都存储在ConcurrentHashMap里面

/** * Store of data * * @author nkorange * @since 1.0.0 */@Componentpublic class DataStore { private Map<String, Datum> dataMap = new ConcurrentHashMap<>(1024); public void put(String key, Datum value) {  dataMap.put(key, value); } public Datum remove(String key) {  return dataMap.remove(key); } public Set<String> keys() {  return dataMap.keySet(); } public Datum get(String key) {  return dataMap.get(key); } public boolean contains(String key) {  return dataMap.containsKey(key); } public Map<String, Datum> batchGet(List<String> keys) {  Map<String, Datum> map = new HashMap<>(128);  for (String key : keys) {   Datum datum = dataMap.get(key);   if (datum == null) {    continue;   }   map.put(key, datum);  }  return map; } public int getInstanceCount() {  int count = 0;  for (Map.Entry<String, Datum> entry : dataMap.entrySet()) {   try {    Datum instancesDatum = entry.getValue();    if (instancesDatum.value instanceof Instances) {     count += ((Instances) instancesDatum.value).getInstanceList().size();    }   } catch (Exception ignore) {   }  }  return count; } public Map<String, Datum> getDataMap() {  return dataMap; }}

　　nacos启动首先从其他节点同步全部数据
DistroConsistencyServiceImpl类构造完成后启动同步线程直到首次同步成功

@org.springframework.stereotype.Service("distroConsistencyService")public class DistroConsistencyServiceImpl implements EphemeralConsistencyService { private volatile Notifier notifier = new Notifier(); private LoadDataTask loadDataTask = new LoadDataTask(); @PostConstruct public void init() {  GlobalExecutor.submit(loadDataTask);  GlobalExecutor.submitDistroNotifyTask(notifier); } /** * 从其它节点同步数据任务线程 */ private class LoadDataTask implements Runnable {  @Override  public void run() {   try {    load();    if (!initialized) {     GlobalExecutor.submit(this, globalConfig.getLoadDataRetryDelayMillis());    }   } catch (Exception e) {    Loggers.DISTRO.error("load data failed.", e);   }  } } /**  * 初始化时从其它节点同步数据  * @throws Exception  */ public void load() throws Exception {  //如果单列模式则返回  if (SystemUtils.STANDALONE_MODE) {   initialized = true;   return;  }  // size = 1 means only myself in the list, we need at least one another server alive:  while (serverListManager.getHealthyServers().size() <= 1) {   Thread.sleep(1000L);   Loggers.DISTRO.info("waiting server list init...");  }  //只要有一个节点同步成功则返回  for (Server server : serverListManager.getHealthyServers()) {   if (NetUtils.localServer().equals(server.getKey())) {    continue;   }   // try sync data from remote server:   if (syncAllDataFromRemote(server)) {    initialized = true;    return;   }  } } /**  * 从指定nacos节点同步数据  * @param server 目标nacos节点  * @return 是否同步成功  */ public boolean syncAllDataFromRemote(Server server) {  try {   byte[] data = NamingProxy.getAllData(server.getKey());   processData(data);   return true;  } catch (Exception e) {   return false;  } }  /**  * 处理同步报文信息  * 更新数据存储和通知监听器  * @param data  * @param data  * @throws Exception  */ public void processData(byte[] data) throws Exception {  if (data.length > 0) {   Map<String, Datum<Instances>> datumMap = serializer.deserializeMap(data, Instances.class);   for (Map.Entry<String, Datum<Instances>> entry : datumMap.entrySet()) {        dataStore.put(entry.getKey(), entry.getValue());    if (!listeners.containsKey(entry.getKey())) {     // pretty sure the service not exist:     if (switchDomain.isDefaultInstanceEphemeral()) {      // create empty service      Loggers.DISTRO.info("creating service {}", entry.getKey());      Service service = new Service();      String serviceName = KeyBuilder.getServiceName(entry.getKey());      String namespaceId = KeyBuilder.getNamespace(entry.getKey());      service.setName(serviceName);      service.setNamespaceId(namespaceId);      service.setGroupName(Constants.DEFAULT_GROUP);      // now validate the service. if failed, exception will be thrown      service.setLastModifiedMillis(System.currentTimeMillis());      service.recalculateChecksum();      listeners.get(KeyBuilder.SERVICE_META_KEY_PREFIX).get(0).onChange(KeyBuilder.buildServiceMetaKey(namespaceId, serviceName), service);     }    }   }   for (Map.Entry<String, Datum<Instances>> entry : datumMap.entrySet()) {    if (!listeners.containsKey(entry.getKey())) {     // Should not happen:     continue;    }    try {     for (RecordListener listener : listeners.get(entry.getKey())) {      listener.onChange(entry.getKey(), entry.getValue().value);     }    } catch (Exception e) {     continue;    }    // Update data store if listener executed successfully:    dataStore.put(entry.getKey(), entry.getValue());   }  } }}public class NamingProxy { /**  * 获取指定nacos节点所有数据  * @param server 目标nacos节点  * @return  返回到数据  * @throws Exception  */ public static byte[] getAllData(String server) throws Exception {  Map<String, String> params = new HashMap<>(8);  /  HttpClient.HttpResult result = HttpClient.httpGet("http://" + server + RunningConfig.getContextPath() + UtilsAndCommons.NACOS_NAMING_CONTEXT + ALL_DATA_GET_URL, new ArrayList<>(), params);  if (HttpURLConnection.HTTP_OK == result.code) {   return result.content.getBytes();  }  throw new IOException("failed to req API: " + "http://" + server + RunningConfig.getContextPath() + UtilsAndCommons.NACOS_NAMING_CONTEXT + DATA_GET_URL + ". code: " + result.code + " msg: " + result.content); }}

Distro两种数据的同步方式：1.定时同步 2.本节点接收到服务变更同步
DataSyncer启动定时同步线程和负责数据同步
TaskDispatcher负责接受服务变更后分发到DataSyncer即时同步

@Component@DependsOn("serverListManager")public class DataSyncer { @PostConstruct //启动定时同步的定时任务 public void init() {  startTimedSync(); } public void startTimedSync() {  //每5秒一次  //dataSyncExecutor.scheduleWithFixedDelay(runnable, PARTITION_DATA_TIMED_SYNC_INTERVAL=5000, PARTITION_DATA_TIMED_SYNC_INTERVAL=5000, TimeUnit.MILLISECONDS);  GlobalExecutor.schedulePartitionDataTimedSync(new TimedSync()); } public class TimedSync implements Runnable {  @Override  public void run() {   try {    // send local timestamps to other servers:    Map<String, String> keyChecksums = new HashMap<>(64);    for (String key : dataStore.keys()) {     //只同步自己负责的那部分数据     // int index = healthyList.indexOf(NetUtils.localServer());     // int target = distroHash(serviceName) % healthyList.size();     if (!distroMapper.responsible(KeyBuilder.getServiceName(key))) {      continue;     }     Datum datum = dataStore.get(key);     if (datum == null) {      continue;     }     keyChecksums.put(key, datum.value.getChecksum());    }    if (keyChecksums.isEmpty()) {     return;    }    for (Server member : getServers()) {     if (NetUtils.localServer().equals(member.getKey())) {      continue;     }     //请求路径/nacos/v1/ns/distro/checksum     NamingProxy.syncCheckSums(keyChecksums, member.getKey());    }   } catch (Exception e) {    Loggers.DISTRO.error("timed sync task failed.", e);   }  } }}

TaskDispatcher负责接受服务变更后分发到DataSyncer即时同步

/** * Data sync task dispatcher * * @author nkorange * @since 1.0.0 */@Componentpublic class TaskDispatcher { @Autowired private GlobalConfig partitionConfig; @Autowired private DataSyncer dataSyncer; private List<TaskScheduler> taskSchedulerList = new ArrayList<>(); private final int cpuCoreCount = Runtime.getRuntime().availableProcessors(); @PostConstruct public void init() {  for (int i = 0; i < cpuCoreCount; i++) {   TaskScheduler taskScheduler = new TaskScheduler(i);   taskSchedulerList.add(taskScheduler);   GlobalExecutor.submitTaskDispatch(taskScheduler);  } }  /**  * 接收服务变更后同步任务  * @param key 服务唯一标识  */ public void addTask(String key) {  taskSchedulerList.get(UtilsAndCommons.shakeUp(key, cpuCoreCount)).addTask(key); } public class TaskScheduler implements Runnable {  private int index;  private int dataSize = 0;  private long lastDispatchTime = 0L;  private BlockingQueue<String> queue = new LinkedBlockingQueue<>(128 * 1024);  public TaskScheduler(int index) {   this.index = index;  }  public void addTask(String key) {   queue.offer(key);  }  public int getIndex() {   return index;  }  @Override  public void run() {   List<String> keys = new ArrayList<>();   while (true) {    try {     String key = queue.poll(partitionConfig.getTaskDispatchPeriod(), TimeUnit.MILLISECONDS);     if (dataSyncer.getServers() == null || dataSyncer.getServers().isEmpty()) {      continue;     }     if (StringUtils.isBlank(key)) {      continue;     }     if (dataSize == 0) {      keys = new ArrayList<>();     }     keys.add(key);     dataSize++;     //批量同步； 1 数量达到batchSyncKeyCount=1000，2：距上次同步事件大于taskDispatchPeriod=2000     if (dataSize == partitionConfig.getBatchSyncKeyCount() ||      (System.currentTimeMillis() - lastDispatchTime) > partitionConfig.getTaskDispatchPeriod()) {      for (Server member : dataSyncer.getServers()) {       if (NetUtils.localServer().equals(member.getKey())) {        continue;       }       SyncTask syncTask = new SyncTask();       syncTask.setKeys(keys);       syncTask.setTargetServer(member.getKey());       //分发任务到dataSyncer同步数据       dataSyncer.submit(syncTask, 0);      }      lastDispatchTime = System.currentTimeMillis();      dataSize = 0;     }    } catch (Exception e) {     Loggers.DISTRO.error("dispatch sync task failed.", e);    }   }  } }}@Component@DependsOn("serverListManager")public class DataSyncer { /**  *  * @param task 同步数据  * @param delay 延迟时间  */ public void submit(SyncTask task, long delay) {  //各种监测  ..................   GlobalExecutor.submitDataSync(() -> {   // 1. check the server    ..................    // 2. get the datums by keys and check the datum is empty or not    ..................     byte[] data = serializer.serialize(datumMap);   long timestamp = System.currentTimeMillis();   // DATA_ON_SYNC_URL = "/nacos/v1/ns/distro/datum"   boolean success = NamingProxy.syncData(data, task.getTargetServer());   ..............   //不成功重试  }, delay); }}

原文转载：http://www.shaoqun.com/a/489725.html

grab：https://www.ikjzd.com/w/841

笨鸟转运：https://www.ikjzd.com/w/1550

美森：https://www.ikjzd.com/w/1693


一致性协议算法Distro阿里自己的创的算法吧，网上能找到的资料很少。Distro用于处理ephemeral类型数据Distro协议算法看代码大体流程是：nacos启动首先从其他远程节点同步全部数据nacos每个节点是平等的都可以处理写入请求，同时把新数据同步到其他节点每个节点只负责部分数据，定时发送自己负责数据校验值到其他节点来保持数据一致性Distro代码分析，nacos负责数据一致性的服务是
法瑞儿：https://www.ikjzd.com/w/412
cares：https://www.ikjzd.com/w/1302
2020年春节深圳欢乐谷有什么优惠活动？：http://tour.shaoqun.com/a/39761.html
亚马逊联手拼多多，是长远谋划还是无奈之举？：https://www.ikjzd.com/home/111600
云南普洱里坎瀑布自驾游怎么走？：http://tour.shaoqun.com/a/43122.html
发帖者 时间：