06篇 Nacos Client本地缓存及故障转移

た入场券 2022-09-05 12:46 241阅读 0赞

> 学习不用那么功利，二师兄带你从更高维度轻松阅读源码～

本篇文章我们来通过源码分析一下Nacos的本地缓存及故障转移功能，涉及到核心类为ServiceInfoHolder和FailoverReactor。

## ServiceInfoHolder功能概述 ##

ServiceInfoHolder类，顾名思义，服务信息的持有者。前面文章已经多次涉及到ServiceInfoHolder类，比如每次客户端从注册中心获取新的服务信息时都会调用该类的processServiceInfo方法来进行本地化的处理，包括更新缓存服务、发布事件、更新本地文件等。

![nacos][]

除了上述功能，该类在实例化时，还做了包含本地缓存目录初始化、故障转移初始化等操作。下面我们就逐一分析一下。

## ServiceInfo的本地内存缓存 ##

ServiceInfo，注册服务的信息，其中包含了服务名称、分组名称、集群信息、实例列表信息、上次更新时间等。也就是说，客户端从注册中心获取到的信息在本地都以ServiceInfo作为承载着。

而ServiceInfoHolder类又持有了ServiceInfo，通过一个ConcurrentMap来存储：

public class ServiceInfoHolder implements Closeable {
        private final ConcurrentMap<String, ServiceInfo> serviceInfoMap;
    }

这就是Nacos客户端对服务注册信息的第一层缓存。前面分析processServiceInfo方法时，我们已经看到，当服务信息变更时会第一时间更新serviceInfoMap中的信息。

public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {
    // ....
        // 缓存服务信息
        serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);
        // 判断注册的实例信息是否已变更
        boolean changed = isChangedServiceInfo(oldService, serviceInfo);
        if (StringUtils.isBlank(serviceInfo.getJsonFromServer())) {
            serviceInfo.setJsonFromServer(JacksonUtils.toJson(serviceInfo));
        }
     // ....
    }

关于serviceInfoMap的使用就这么简单，当变动实例向其中put最新数据即可。当使用实例，根据key进行get操作即可。

而serviceInfoMap在ServiceInfoHolder的构造方法中进行初始化，默认创建一个空的ConcurrentMap。但当配置了启动时从缓存文件读取信息时，则会从本地缓存进行加载。

// 启动时是否从缓存目录读取信息，默认false。设置为true会读取缓存文件
    if (isLoadCacheAtStart(properties)) {
        this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(DiskCache.read(this.cacheDir));
    } else {
        this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(16);
    }

这里涉及到了本地缓存目录，在processServiceInfo方法中，当服务实例变更时，会看到通过DiskCache\#write方法向该目录写入ServiceInfo信息。

// 服务实例已变更
    if (changed) {
        NAMING_LOGGER.info("current ips:(" + serviceInfo.ipCount() + ") service: " + serviceInfo.getKey() + " -> "
                + JacksonUtils.toJson(serviceInfo.getHosts()));
        // 添加实例变更事件，会被推动到订阅者执行
        NotifyCenter.publishEvent(new InstancesChangeEvent(serviceInfo.getName(), serviceInfo.getGroupName(),
                serviceInfo.getClusters(), serviceInfo.getHosts()));
        // 记录Service本地文件
        DiskCache.write(serviceInfo, cacheDir);
    }

下面就来聊聊本地缓存目录。

## 本地缓存目录 ##

本地缓存目录作为ServiceInfoHolder的一个属性存在，用于指定本地缓存的根目录和故障转移的根目录。

private String cacheDir;

在ServiceInfoHolder的构造方法中，第一个调用的便是生成缓存目录：

public ServiceInfoHolder(String namespace, Properties properties) {
        // 生成缓存目录：默认为${user.home}/nacos/naming/public，
        // 可以通过System.setProperty("JM.SNAPSHOT.PATH")自定义根目录
        initCacheDir(namespace, properties);
        //...
    }

关于生成目录的源码就不看了，默认缓存目录为$\{user.home\}/nacos/naming/public，可以通过System.setProperty(“JM.SNAPSHOT.PATH”)自定义根目录。

初始化完该目录之后，故障转移信息也存储在该目录下。

## 故障转移 ##

同样在ServiceInfoHolder的构造方法中，会初始化一个FailoverReactor类，同样是ServiceInfoHolder的成员变量。FailoverReactor的作用便是用来处理故障转移的。

this.failoverReactor = new FailoverReactor(this, cacheDir);

这里的this为ServiceInfoHolder当前的对象，也就是说两者相互持有对方的引用。

来看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的构造方法基本上把它的功能都展示出来了：

*  持有ServiceInfoHolder引用；
 *  拼接故障根目录：$\{user.home\}/nacos/naming/public/failover，其中public也有可能是其他的自定义命名空间；
 *  初始化executorService；
 *  init方法：通过executorService开启多个定时任务执行；

### init方法执行 ###

init方法中开启了三个定时任务：

*  初始化立即执行，执行间隔5秒，执行任务为SwitchRefresher；
 *  初始化延迟30分钟执行，执行间隔24小时，执行任务为DiskFileWriter；
 *  初始化立即执行，执行间隔10秒，执行核心操作为DiskFileWriter；

这三个任务都是FailoverReactor的内部类，先看后两个任务DiskFileWriter的实现：

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();
                if (StringUtils.equals(serviceInfo.getKey(), UtilAndComs.ALL_IPS) || StringUtils
                        .equals(serviceInfo.getName(), UtilAndComs.ENV_LIST_KEY) || StringUtils
                        .equals(serviceInfo.getName(), UtilAndComs.ENV_CONFIGS) || StringUtils
                        .equals(serviceInfo.getName(), UtilAndComs.VIP_CLIENT_FILE) || StringUtils
                        .equals(serviceInfo.getName(), UtilAndComs.ALL_HOSTS)) {
                    continue;
                }
                // 将缓存内容写入磁盘文件
                DiskCache.write(serviceInfo, failoverDir);
            }
        }
    }

逻辑非常简单，就是获取ServiceInfoHolder中缓存的ServiceInfo，判断是否满足写入磁盘文件，如果满足，则将其写入前面拼接的故障转移目录：$\{user.home\}/nacos/naming/public/failover。只不过第二个定时任务和第三个定时任务的区别时，第三个定时任务有前置判断，只有当文件不存在时才执行。

最后再来看一下SwitchRefresher的核心实现如下：

File switchFile = new File(failoverDir + UtilAndComs.FAILOVER_SWITCH);
    // 文件不存在退出
    if (!switchFile.exists()) {
        switchParams.put("failover-mode", "false");
        NAMING_LOGGER.debug("failover switch is not found, " + switchFile.getName());
        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());
                    NAMING_LOGGER.info("failover-mode is on");
                    new FailoverFileReader().run();
                } else if (NO_FAILOVER_MODE.equals(line1)) {
                    // 0表示关闭故障转移模式
                    switchParams.put(FAILOVER_MODE_PARAM, Boolean.FALSE.toString());
                    NAMING_LOGGER.info("failover-mode is off");
                }
            }
        } else {
            switchParams.put(FAILOVER_MODE_PARAM, Boolean.FALSE.toString());
        }
    }

上述代码的逻辑梳理如下：

*  如果故障转移文件不存在，则直接返回。故障转移【开关】文件为名为“00-00—000-VIPSRV\_FAILOVER\_SWITCH-000—00-00”。
 *  比较文件修改时间，如果已经修改，则获取故障转移文件中的内容。
 *  故障转移文件中存储了0和1标识。0表示关闭，1表示开启。
 *  当为开启状态时，执行线程FailoverFileReader。

FailoverFileReader，顾名思义，就是故障转移文件读取。基本操作就是读取failover目录存储ServiceInfo的文件内容，然后转换成ServiceInfo，并用将所有的ServiceInfo存储在FailoverReactor的serviceMap属性中。

failover目录文件内容示例如下：

(base) appledeMacBook-Pro-2:failover apple$ ls
    DEFAULT_GROUP%40%40nacos.test.1
    DEFAULT_GROUP%40%40user-provider@@DEFAULT
    DEFAULT_GROUP%40%40user-service-consumer@@DEFAULT
    DEFAULT_GROUP%40%40user-service-provider
    DEFAULT_GROUP%40%40user-service-provider@@DEFAULT

文件内容格式如下：

{
        "hosts": [
            {
                "ip": "1.1.1.1",
                "port": 800,
                "valid": true,
                "healthy": true,
                "marked": false,
                "instanceId": "1.1.1.1#800#DEFAULT#DEFAULT_GROUP@@nacos.test.1",
                "metadata": {
                    "netType": "external",
                    "version": "2.0"
                },
                "enabled": true,
                "weight": 2,
                "clusterName": "DEFAULT",
                "serviceName": "DEFAULT_GROUP@@nacos.test.1",
                "ephemeral": true
            }
        ],
        "dom": "DEFAULT_GROUP@@nacos.test.1",
        "name": "DEFAULT_GROUP@@nacos.test.1",
        "cacheMillis": 10000,
        "lastRefTime": 1617001291656,
        "checksum": "969c531798aedb72f87ac686dfea2569",
        "useSpecifiedURL": false,
        "clusters": "",
        "env": "",
        "metadata": {}
    }

下面看一下其中的核心业务实现：

for (File file : files) {
        if (!file.isFile()) {
            continue;
        }
    
        // 如果是故障转移标志文件，则跳过
        if (file.getName().equals(UtilAndComs.FAILOVER_SWITCH)) {
            continue;
        }
    
        ServiceInfo dom = new ServiceInfo(file.getName());
    
        try {
            String dataString = ConcurrentDiskUtil
                    .getFileContent(file, Charset.defaultCharset().toString());
            reader = new BufferedReader(new StringReader(dataString));
    
            String json;
            if ((json = reader.readLine()) != null) {
                try {
                    dom = JacksonUtils.toObj(json, ServiceInfo.class);
                } catch (Exception e) {
                    NAMING_LOGGER.error("[NA] error while parsing cached dom : " + json, e);
                }
            }
    
        } catch (Exception e) {
            NAMING_LOGGER.error("[NA] failed to read cache for dom: " + file.getName(), e);
        } finally {
            try {
                if (reader != null) {
                    reader.close();
                }
            } catch (Exception e) {
                //ignore
            }
        }
    
        // ... 读入缓存
        if (!CollectionUtils.isEmpty(dom.getHosts())) {
            domMap.put(dom.getKey(), dom);
        }
    }

代码基本流程如下：

*  读取failover目录下的所有文件，进行遍历处理；
 *  如果文件不存在，跳过；
 *  如果文件是故障转移标志文件，跳过；
 *  读取文件中的json内容，转化为ServiceInfo对象；
 *  将ServiceInfo对象放入domMap当中；

当for循环执行完毕，如果domMap不为空，则将其赋值给serviceMap：

if (domMap.size() > 0) {
        serviceMap = domMap;
    }

那么，有同学会问了，这个serviceMap在哪里用到呢？前面我们讲获取实例的时候，通常会调用一个名为getServiceInfo的方法：

public ServiceInfo getServiceInfo(final String serviceName, final String groupName, final String clusters) {
        NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch());
        String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
        String key = ServiceInfo.getKey(groupedServiceName, clusters);
        if (failoverReactor.isFailoverSwitch()) {
            return failoverReactor.getService(key);
        }
        return serviceInfoMap.get(key);
    }

也就是说，如果开启了故障转移，则会优先调用failoverReactor\#getService方法，而这个方法便是从serviceMap中获取ServiceInfo。

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

至此，关于Nacos客户端的故障转移流程分析完毕。

## 小结 ##

本篇文章介绍了Nacos客户端本地缓存及故障转移的实现。所谓的本地缓存有两方面，第一方面是从注册中心获得实例信息会缓存在内存当中，也就是通过Map的形式承载，这样查询操作都方便。第二方法便是通过磁盘文件的形式定时缓存起来，以备不时之需。

而故障转移也分两方面，第一方面是故障转移的开关是通过文件来标记的；第二方面是当开启故障转移之后，当发生故障时，可以从故障转移定时备份的文件中来获得服务实例信息。

> 博主简介：《SpringBoot技术内幕》技术图书作者，酷爱钻研技术，写技术干货文章。
> 
> 公众号：「程序新视界」，博主的公众号，欢迎关注~
> 
> 技术交流：请联系博主微信号：zhuan2quan

[nacos]: /images/20220829/551c9c3dc60e4a91ab02649f7328fa0b.png