Nacos客户端本地缓存和故障转移方式

// ServiceInfoHolder
private final ConcurrentMap<String, ServiceInfo> serviceInfoMap;

// ServiceInfoHolder
public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {
    ....
    //缓存服务信息
    serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);
    ....
}

// ServiceInfoHolder
public ServiceInfoHolder(String namespace, Properties properties) {
    initCacheDir(namespace, properties);
    if (isLoadCacheAtStart(properties)) {
        this.serviceInfoMap = new ConcurrentHashMap<>(DiskCache.read(this.cacheDir));
    } else {
        this.serviceInfoMap = new ConcurrentHashMap<>(16);
    }
    this.failoverReactor = new FailoverReactor(this, cacheDir);
    this.pushEmptyProtection = isPushEmptyProtect(properties);
}

// ServiceInfoHolder
public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {
    ....
    // 记录Service本地文件
    DiskCache.write(serviceInfo, cacheDir);
    ....
}

// FailoverReactor
public FailoverReactor(ServiceInfoHolder serviceInfoHolder, String cacheDir) {
    // 持有ServiceInfoHolder的引用
    this.serviceInfoHolder = serviceInfoHolder;
    // 拼接故障目录：${user.home}/nacos/naming/public/failover
    this.failoverDir = cacheDir + FAILOVER_DIR;
    // 初始化executorService，支持延时执行
    this.executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            // 守护线程模式运行
            thread.setDaemon(true);
            thread.setName("com.alibaba.nacos.naming.failover");
            return thread;
        }
    });
    // 其他初始化操作，通过executorService开启多个定时任务执行
    this.init();
}

// FailoverReactor
public void init() {
    // 初始化立即执行，执行间隔5秒，执行任务SwitchRefresher
    executorService.scheduleWithFixedDelay(new SwitchRefresher(), 0L, 5000L, TimeUnit.MILLISECONDS);
    // 初始化延迟30分钟执行，执行间隔24小时，执行任务DiskFileWriter
    executorService.scheduleWithFixedDelay(new DiskFileWriter(), 30, DAY_PERIOD_MINUTES, TimeUnit.MINUTES);
    // 10秒后如果故障目录为空，则执行DiskFileWriter任务强制备份
    executorService.schedule(new Runnable() {
        @Override
        public void run() {
            try {
                File cacheDir = new File(failoverDir);
                ...
                File[] files = cacheDir.listFiles();
                if (files == null || files.length <= 0) {
                    new DiskFileWriter().run();
                }
            } catch (Throwable e) {
                NAMING_LOGGER.error("[NA] failed to backup file on startup.", e);
            }

        }
    }, 10000L, TimeUnit.MILLISECONDS);
}

// FailoverReactor
class DiskFileWriter extends TimerTask {
    @Override
    public void run() {
        Map<String, ServiceInfo> map = serviceInfoHolder.getServiceInfoMap();
        for (Map.Entry<String, ServiceInfo> entry : map.entrySet()) {
            ServiceInfo serviceInfo = entry.getValue();
            ...
            // 将缓存写入磁盘
            DiskCache.write(serviceInfo, failoverDir);
        }
    }
}

// FailoverReactor
class SwitchRefresher implements Runnable {
    long lastModifiedMillis = 0L;
    @Override
    public void run() {
        try {
            File switchFile = new File(failoverDir + UtilAndComs.FAILOVER_SWITCH);
            // 文件不存在则退出
            if (!switchFile.exists()) {
                ...
                return;
            }

            long modified = switchFile.lastModified();
      
            if (lastModifiedMillis < modified) {
                lastModifiedMillis = modified;
                // 获取故障转移文件内容
                String failover = ConcurrentDiskUtil.getFileContent(failoverDir + UtilAndComs.FAILOVER_SWITCH,
                                                                    Charset.defaultCharset().toString());
                if (!StringUtils.isEmpty(failover)) {
                    String[] lines = failover.split(DiskCache.getLineSeparator());

                    for (String line : lines) {
                        String line1 = line.trim();
                        // 1 表示开启故障转移模式
                        if (IS_FAILOVER_MODE.equals(line1)) {
                            switchParams.put(FAILOVER_MODE_PARAM, Boolean.TRUE.toString());
                            new FailoverFileReader().run();
                        // 0 表示关闭故障转移模式
                        } else if (NO_FAILOVER_MODE.equals(line1)) {
                            switchParams.put(FAILOVER_MODE_PARAM, Boolean.FALSE.toString());
                        }
                    }
                } else {
                    switchParams.put(FAILOVER_MODE_PARAM, Boolean.FALSE.toString());
                }
            }

        } catch (Throwable e) {
            NAMING_LOGGER.error("[NA] failed to read failover switch.", e);
        }
    }
}

// FailoverReactor
class FailoverFileReader implements Runnable {
    @Override
    public void run() {
        Map<String, ServiceInfo> domMap = new HashMap<String, ServiceInfo>(16);
        File cacheDir = new File(failoverDir);
        File[] files = cacheDir.listFiles();
        if (files == null) {
            return;
        }
        for (File file : files) {
            // 如果是故障转移标志文件，则跳过
            if (file.getName().equals(UtilAndComs.FAILOVER_SWITCH)) {
                continue;
            }
            ServiceInfo dom = new ServiceInfo(file.getName());
            BufferedReader reader = null;
            try {
                String dataString = ConcurrentDiskUtil
                    .getFileContent(file, Charset.defaultCharset().toString());
                reader = new BufferedReader(new StringReader(dataString));
                String json = reader.readLine();
                dom = JacksonUtils.toObj(json, ServiceInfo.class);
            } finally {
                reader.close();
            }
            if (!CollectionUtils.isEmpty(dom.getHosts())) {
                domMap.put(dom.getKey(), dom);
            }
        }
        // 读入缓存
        if (domMap.size() > 0) {
            serviceMap = domMap;
        }
    }
}·

// ServiceInfoHolder
public ServiceInfo getServiceInfo(final String serviceName, final String groupName, final String clusters) {
    ...
    if (failoverReactor.isFailoverSwitch()) {
        return failoverReactor.getService(key);
    }
    return serviceInfoMap.get(key);
}

// FailoverReactor
public ServiceInfo getService(String key) {
    ServiceInfo serviceInfo = serviceMap.get(key);
    if (serviceInfo == null) {
        serviceInfo = new ServiceInfo();
        serviceInfo.setName(key);
    }
    return serviceInfo;
}