Spring Cloud Alibaba——Nacos AP一致性策略 Distro

Eureka 一致性策略

Eureka是一个AP模式的服务发现框架，在Eureka集群模式下，Eureka采取的是Server之间互相广播各自的数据进行数据复制、更新操作；并且Eureka在客户端与注册中心出现网络故障时，依然能够获取服务注册信息——Eureka实现了客户端对于服务注册信息的缓存

@Singleton public class DiscoveryClient implements EurekaClient { private boolean fetchRegistryFromBackup() { try { @SuppressWarnings("deprecation") BackupRegistry backupRegistryInstance = newBackupRegistryInstance(); if (null == backupRegistryInstance) { // backward compatibility with the old protected method, in case it is being used. backupRegistryInstance = backupRegistryProvider.get(); } if (null != backupRegistryInstance) { Applications apps = null; if (isFetchingRemoteRegionRegistries()) { String remoteRegionsStr = remoteRegionsToFetch.get(); if (null != remoteRegionsStr) { apps = backupRegistryInstance.fetchRegistry(remoteRegionsStr.split(",")); } } else { apps = backupRegistryInstance.fetchRegistry(); } if (apps != null) { final Applications applications = this.filterAndShuffle(apps); applications.setAppsHashCode(applications.getReconcileHashCode()); localRegionApps.set(applications); logTotalInstances(); logger.info("Fetched registry successfully from the backup"); return true; } } else { logger.warn("No backup registry instance defined & unable to find any discovery servers."); } } catch (Throwable e) { logger.warn("Cannot fetch applications from apps although backup registry was specified", e); } return false; } }

正因为Eureka为了能够在Eureka集群无法工作时不影响消费者调用服务提供者而设置的客户端缓存，因此Eureka无法保证服务注册信息的强一致性（CP模式），只能满足数据的最终一致性（AP模式）

Nacos AP一致性策略——Distro

Nacos在AP模式下的一致性策略就类似于Eureka，采用Server之间互相的数据同步来实现数据在集群中的同步、复制操作。

触发数据广播

@DependsOn("ProtocolManager") @org.springframework.stereotype.Service("distroConsistencyService") public class DistroConsistencyServiceImpl implements EphemeralConsistencyService, DistroDataProcessor { private final GlobalConfig globalConfig; private final DistroProtocol distroProtocol; @Override 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); } }

当服务注册和注销实例，即InstanceController的register和deregister方法被调用时，ServiceManager类中addInstance或removeInstance方法调用ConsistencyService接口定义的put、remove方法时，涉及到了Server端数据的变更，此时会创建一个任务DistroDelayTask，将数据的key封装到DistroDelayTask中。

@Component public class DistroProtocol { public void sync(DistroKey distroKey, DataOperation action, long delay) { for (Member each : memberManager.allMembersWithoutSelf()) { //遍历所有服务ip。进行数据同步 DistroKey distroKeyWithTarget = new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(), each.getAddress()); //创建延迟任务 DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay); //其实也是一个定时任务 distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget, distroDelayTask); if (Loggers.DISTRO.isDebugEnabled()) { Loggers.DISTRO.debug("[DISTRO-SCHEDULE] {} to {}", distroKey, each.getAddress()); } } } }

然后将DistroDelayTask提交到Nacos的延时任务执行引擎NacosDelayTaskExecuteEngine中的ConcurrentHashMap<Object, AbstractDelayTask> tasks中存储，并且NacosDelayTaskExecuteEngine在其构造方法中初始化ScheduledExecutorService线程池并提交一个ProcessRunnable任务去取出tasks中的AbstractDelayTask任务进行处理。

public class NacosDelayTaskExecuteEngine extends AbstractNacosTaskExecuteEngine<AbstractDelayTask> { private final ScheduledExecutorService processingExecutor; protected final ConcurrentHashMap<Object, AbstractDelayTask> tasks; protected final ReentrantLock lock = new ReentrantLock(); //NacosDelayTaskExecuteEngine实例化 public NacosDelayTaskExecuteEngine(String name, int initCapacity, Logger logger, long processInterval) { super(logger); //初始化tasks tasks = new ConcurrentHashMap<Object, AbstractDelayTask>(initCapacity); //创建定时任务线程池 processingExecutor = ExecutorFactory.newSingleScheduledExecutorService(new NameThreadFactory(name)); //每间隔一段时间定时执行任务ProcessRunnable processingExecutor .scheduleWithFixedDelay(new ProcessRunnable(), processInterval, processInterval, TimeUnit.MILLISECONDS); } @Override public void addTask(Object key, AbstractDelayTask newTask) { lock.lock(); try { AbstractDelayTask existTask = tasks.get(key); //存在相同任务则合并 if (null != existTask) { newTask.merge(existTask); } //将AbstractDelayTask存入tasks中 tasks.put(key, newTask); } finally { lock.unlock(); } }

ProcessRunnable线程处理存入延时任务执行引擎NacosDelayTaskExecuteEngine中的ConcurrentHashMap<Object, AbstractDelayTask> tasks中的任务

public class NacosDelayTaskExecuteEngine extends AbstractNacosTaskExecuteEngine<AbstractDelayTask> { private final ScheduledExecutorService processingExecutor; protected final ConcurrentHashMap<Object, AbstractDelayTask> tasks; /** * process tasks in execute engine. */ protected void processTasks() { //获取tasks中的所有key Collection<Object> keys = getAllTaskKeys(); //遍历任务key集合 for (Object taskKey : keys) { //删除tasks中的AbstractDelayTask任务 AbstractDelayTask task = removeTask(taskKey); if (null == task) { continue; } //获取NacosTaskProcessor,这里实质获取的是DistroDelayTaskProcessor //在DistroHttpRegistry中的doRegister方法中提前设置进去了DistroHttpDelayTaskProcessor //但是taskKey最为key会取不到DistroHttpDelayTaskProcessor //而DistroTaskEngineHolder类的构造函数中设置了DistroDelayTaskExecuteEngine类 //中的DefaultTaskProcessor为DistroDelayTaskProcessor NacosTaskProcessor processor = getProcessor(taskKey); if (null == processor) { getEngineLog().error("processor not found for task, so discarded. " + task); continue; } try { // ReAdd task if process failed //DistroHttpDelayTaskProcessor执行AbstractDelayTask任务 if (!processor.process(task)) { //执行失败，更新最后执行时间，并将该任务重新设置进tasks中，便于后续继续执行 retryFailedTask(taskKey, task); } } catch (Throwable e) { getEngineLog().error("Nacos task execute error : " + e.toString(), e); //执行异常，更新最后执行时间，并将该任务重新设置进tasks中，便于后续继续执行 retryFailedTask(taskKey, task); } } } private void retryFailedTask(Object key, AbstractDelayTask task) { //更新最后执行时间 task.setLastProcessTime(System.currentTimeMillis()); //并将该任务重新设置进tasks中，便于后续继续执行 addTask(key, task); } private class ProcessRunnable implements Runnable { @Override public void run() { try { //处理任务 processTasks(); } catch (Throwable e) { getEngineLog().error(e.toString(), e); } } } }

在DistroTaskEngineHolder类的构造函数中设置了DistroDelayTaskExecuteEngine类中的DefaultTaskProcessor为DistroDelayTaskProcessor

public abstract class AbstractNacosTaskExecuteEngine<T extends NacosTask> implements NacosTaskExecuteEngine<T> { private final ConcurrentHashMap<Object, NacosTaskProcessor> taskProcessors = new ConcurrentHashMap<Object, NacosTaskProcessor>(); private NacosTaskProcessor defaultTaskProcessor; @Override public NacosTaskProcessor getProcessor(Object key) { //获取任务处理器 return taskProcessors.containsKey(key) ? taskProcessors.get(key) : defaultTaskProcessor; } } @Component public class DistroHttpRegistry { private final DistroComponentHolder componentHolder; private final DistroTaskEngineHolder taskEngineHolder; @PostConstruct public void doRegister() { componentHolder.registerDataStorage(KeyBuilder.INSTANCE_LIST_KEY_PREFIX, new DistroDataStorageImpl(dataStore, distroMapper)); componentHolder.registerTransportAgent(KeyBuilder.INSTANCE_LIST_KEY_PREFIX, new DistroHttpAgent(memberManager)); componentHolder.registerFailedTaskHandler(KeyBuilder.INSTANCE_LIST_KEY_PREFIX, new DistroHttpCombinedKeyTaskFailedHandler(globalConfig, taskEngineHolder)); //DistroHttpRegistry实例化后设置了DistroHttpDelayTaskProcessor taskEngineHolder.registerNacosTaskProcessor(KeyBuilder.INSTANCE_LIST_KEY_PREFIX, new DistroHttpDelayTaskProcessor(globalConfig, taskEngineHolder)); componentHolder.registerDataProcessor(consistencyService); } } @Component public class DistroTaskEngineHolder { private final DistroDelayTaskExecuteEngine delayTaskExecuteEngine = new DistroDelayTaskExecuteEngine(); private final DistroExecuteTaskExecuteEngine executeWorkersManager = new DistroExecuteTaskExecuteEngine(); public DistroTaskEngineHolder(DistroComponentHolder distroComponentHolder) { DistroDelayTaskProcessor defaultDelayTaskProcessor = new DistroDelayTaskProcessor(this, distroComponentHolder); //设置了DistroDelayTaskExecuteEngine类中的DefaultTaskProcessor为DistroDelayTaskProcessor delayTaskExecuteEngine.setDefaultTaskProcessor(defaultDelayTaskProcessor); } }

DistroDelayTaskProcessor执行AbstractDelayTask任务

• 在DistroDelayTaskProcessor中process方法中创建了DistroSyncChangeTask任务

public class DistroDelayTaskProcessor implements NacosTaskProcessor { private final DistroTaskEngineHolder distroTaskEngineHolder; @Override public boolean process(NacosTask task) { if (!(task instanceof DistroDelayTask)) { return true; } DistroDelayTask distroDelayTask = (DistroDelayTask) task; DistroKey distroKey = distroDelayTask.getDistroKey(); if (DataOperation.CHANGE.equals(distroDelayTask.getAction())) { //又创建了一个定时任务 DistroSyncChangeTask syncChangeTask = new DistroSyncChangeTask(distroKey, distroComponentHolder); distroTaskEngineHolder.getExecuteWorkersManager().addTask(distroKey, syncChangeTask); return true; } return false; } }

DistroSyncChangeTask任务会被添加进TaskExecuteWorker持有的BlockingQueue队列中用于异步执行。

TaskExecuteWorker中的构造函数开启InnerWorker线程，实时监听queue中的任务，进行消费。

public class NacosExecuteTaskExecuteEngine extends AbstractNacosTaskExecuteEngine<AbstractExecuteTask> { private final TaskExecuteWorker[] executeWorkers; @Override public void addTask(Object tag, AbstractExecuteTask task) { NacosTaskProcessor processor = getProcessor(tag); if (null != processor) { processor.process(task); return; } TaskExecuteWorker worker = getWorker(tag); //执行任务 worker.process(task); } } public final class TaskExecuteWorker implements NacosTaskProcessor, Closeable { /** * Max task queue size 32768. 队列大小 */ private static final int QUEUE_CAPACITY = 1 << 15; private final Logger log; private final String name; private final BlockingQueue<Runnable> queue; public TaskExecuteWorker(final String name, final int mod, final int total, final Logger logger) { this.name = name + "_" + mod + "%" + total; //初始化队列 this.queue = new ArrayBlockingQueue<Runnable>(QUEUE_CAPACITY); this.closed = new AtomicBoolean(false); this.log = null == logger ? LoggerFactory.getLogger(TaskExecuteWorker.class) : logger; //开启InnerWorker线程，实时监听queue中的任务，进行消费 new InnerWorker(name).start(); } @Override public boolean process(NacosTask task) { //DistroSyncChangeTask 是AbstractExecuteTask的子类 if (task instanceof AbstractExecuteTask) { //添加任务 putTask((Runnable) task); } return true; } private void putTask(Runnable task) { try { //向队列中添加任务 queue.put(task); } catch (InterruptedException ire) { log.error(ire.toString(), ire); } } }

TaskExecuteWorker中的构造函数开启InnerWorker线程，实时监听queue中的任务，进行消费，执行任务。

public final class TaskExecuteWorker implements NacosTaskProcessor, Closeable { private final BlockingQueue<Runnable> queue; public TaskExecuteWorker(final String name, final int mod, final int total, final Logger logger) { this.name = name + "_" + mod + "%" + total; //初始化队列 this.queue = new ArrayBlockingQueue<Runnable>(QUEUE_CAPACITY); this.closed = new AtomicBoolean(false); this.log = null == logger ? LoggerFactory.getLogger(TaskExecuteWorker.class) : logger; //开启InnerWorker线程，实时监听queue中的任务，进行消费 new InnerWorker(name).start(); } private class InnerWorker extends Thread { InnerWorker(String name) { setDaemon(false); setName(name); } @Override public void run() { while (!closed.get()) { try { //获取队列中的任务 Runnable task = queue.take(); long begin = System.currentTimeMillis(); //执行任务 task.run(); long duration = System.currentTimeMillis() - begin; if (duration > 1000L) { log.warn("distro task {} takes {}ms", task, duration); } } catch (Throwable e) { log.error("[DISTRO-FAILED] " + e.toString(), e); } } } } }

DistroSyncChangeTask线程

• 首先通过DistroKey中的key集合去DataStore中去查询key所对应的数据集合，然后对数据进行序列化操作，转为byte[]数组后，执行Http请求操作——NamingProxy.syncData(data, task.getTargetServer())；如果数据广播失败，则重试失败任务，重新包装为DistroDelayTask任务进行执行。

public class DistroSyncChangeTask extends AbstractDistroExecuteTask { private final DistroComponentHolder distroComponentHolder; @Override public void run() { Loggers.DISTRO.info("[DISTRO-START] {}", toString()); try { String type = getDistroKey().getResourceType(); //key集合去DataStore中去查询key所对应的数据集合 DistroData distroData = distroComponentHolder.findDataStorage(type).getDistroData(getDistroKey()); distroData.setType(DataOperation.CHANGE); //是调用接口地址，通知其他服务,同步服务变更数据 boolean result = distroComponentHolder.findTransportAgent(type).syncData(distroData, getDistroKey().getTargetServer()); if (!result) { //重试失败任务 handleFailedTask(); } Loggers.DISTRO.info("[DISTRO-END] {} result: {}", toString(), result); } catch (Exception e) { Loggers.DISTRO.warn("[DISTRO] Sync data change failed.", e); handleFailedTask(); } } } public class DistroHttpAgent implements DistroTransportAgent { private final ServerMemberManager memberManager; @Override public boolean syncData(DistroData data, String targetServer) { if (!memberManager.hasMember(targetServer)) { return true; } byte[] dataContent = data.getContent(); //同步数据 return NamingProxy.syncData(dataContent, data.getDistroKey().getTargetServer()); } } public class NamingProxy { private static final String DATA_ON_SYNC_URL = "/distro/datum"; public static boolean syncData(byte[] data, String curServer) { Map<String, String> headers = new HashMap<>(128); headers.put(HttpHeaderConsts.CLIENT_VERSION_HEADER, VersionUtils.version); headers.put(HttpHeaderConsts.USER_AGENT_HEADER, UtilsAndCommons.SERVER_VERSION); headers.put(HttpHeaderConsts.ACCEPT_ENCODING, "gzip,deflate,sdch"); headers.put(HttpHeaderConsts.CONNECTION, "Keep-Alive"); headers.put(HttpHeaderConsts.CONTENT_ENCODING, "gzip"); try { //请求/distro/datum接口 RestResult<String> result = HttpClient.httpPutLarge( "http://" + curServer + EnvUtil.getContextPath() + UtilsAndCommons.NACOS_NAMING_CONTEXT + DATA_ON_SYNC_URL, headers, data); if (result.ok()) { return true; } if (HttpURLConnection.HTTP_NOT_MODIFIED == result.getCode()) { return true; } throw new IOException("failed to req API:" + "http://" + curServer + EnvUtil.getContextPath() + UtilsAndCommons.NACOS_NAMING_CONTEXT + DATA_ON_SYNC_URL + ". code:" + result.getCode() + " msg: " + result.getData()); } catch (Exception e) { Loggers.SRV_LOG.warn("NamingProxy", e); } return false; } }

这里将数据提交到了URL为PUT http://ip:port/nacos/v1/ns/distro/datum中，而该URL对应的处理器为DistroController中的public String onSyncDatum(HttpServletRequest request, HttpServletResponse response)方法

@RestController @RequestMapping(UtilsAndCommons.NACOS_NAMING_CONTEXT + "/distro") public class DistroController { @Autowired private DistroProtocol distroProtocol; @Autowired private ServiceManager serviceManager; @PutMapping("/datum") public ResponseEntity onSyncDatum(@RequestBody Map<String, Datum<Instances>> dataMap) throws Exception { if (dataMap.isEmpty()) { Loggers.DISTRO.error("[onSync] receive empty entity!"); throw new NacosException(NacosException.INVALID_PARAM, "receive empty entity!"); } for (Map.Entry<String, Datum<Instances>> entry : dataMap.entrySet()) { if (KeyBuilder.matchEphemeralInstanceListKey(entry.getKey())) { String namespaceId = KeyBuilder.getNamespace(entry.getKey()); String serviceName = KeyBuilder.getServiceName(entry.getKey()); if (!serviceManager.containService(namespaceId, serviceName) && switchDomain .isDefaultInstanceEphemeral()) { serviceManager.createEmptyService(namespaceId, serviceName, true); } DistroHttpData distroHttpData = new DistroHttpData(createDistroKey(entry.getKey()), entry.getValue()); distroProtocol.onReceive(distroHttpData); } } return ResponseEntity.ok("ok"); } }

这里会调用DistroConsistencyServiceImpl.processData(DistroData distroData)方法，然后接着调用DistroConsistencyServiceImpl.onPut(String key, Record value)方法，将数据添加到Notifier中的ArrayBlockingQueue队列tasks中进行服务的注册。

 

Nacos服务端注册流程：https://www.jianshu.com/p/99f67f6f2577

至此完成了Nacos Server在AP模式下的数据的最终一致性操作。

参考： https://www.liaochuntao.cn/2019/05/09/java-web-32/