Spring Cloud Alibaba——Nacos AP一致
2021-07-30 本文已影响0人
小波同学
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模式下的数据的最终一致性操作。
