Nacos客戶端本地緩存和故障轉(zhuǎn)移方式

更新時(shí)間：2024年12月28日 08:48:36 作者：91猿說(shuō)編程

Nacos客戶端在從Server獲得服務(wù)時(shí),若出現(xiàn)故障,會(huì)通過(guò)ServiceInfoHolder和FailoverReactor進(jìn)行故障轉(zhuǎn)移,ServiceInfoHolder緩存服務(wù)信息,FailoverReactor處理故障轉(zhuǎn)移,包括開(kāi)啟故障轉(zhuǎn)移開(kāi)關(guān)、讀取備份文件等

在Nacos客戶端從Server獲得服務(wù)的時(shí)候，在某些時(shí)候出現(xiàn)了一些故障，這時(shí)候?yàn)榱吮ＷC服務(wù)正常，Nacos進(jìn)行了故障轉(zhuǎn)移，原理就是將之前緩存的服務(wù)信息拿出來(lái)用，防止服務(wù)出現(xiàn)問(wèn)題，涉及到的核心類(lèi)為ServiceInfoHolder和FailoverReactor。

本地緩存有兩方面，第一方面是從注冊(cè)中心獲得實(shí)例信息會(huì)緩存在內(nèi)存當(dāng)中，也就是通過(guò)Map的形式承載，這樣查詢操作都方便。第二方面便是通過(guò)磁盤(pán)文件的形式定時(shí)緩存起來(lái)，以備不時(shí)之需。

故障轉(zhuǎn)移也分兩方面，第一方面是故障轉(zhuǎn)移的開(kāi)關(guān)是通過(guò)文件來(lái)標(biāo)記的；第二方面是當(dāng)開(kāi)啟故障轉(zhuǎn)移之后，當(dāng)發(fā)生故障時(shí)，可以從故障轉(zhuǎn)移備份的文件中來(lái)獲得服務(wù)實(shí)例信息。

1. ServiceInfoHolder

ServiceInfoHolder類(lèi)，顧名思義，服務(wù)信息的持有者。每次客戶端從注冊(cè)中心獲取新的服務(wù)信息時(shí)都會(huì)調(diào)用該類(lèi)，其中processServiceInfo方法來(lái)進(jìn)行本地化處理，包括更新緩存服務(wù)、發(fā)布事件、更新本地文件等。

ServiceInfoHolder類(lèi)持有了ServiceInfo，通過(guò)一個(gè)ConcurrentMap來(lái)儲(chǔ)存

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

當(dāng)從服務(wù)端拉會(huì)服務(wù)信息時(shí)，就會(huì)往這個(gè)map中進(jìn)行存儲(chǔ)。

// ServiceInfoHolder
public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {
    ....
    //緩存服務(wù)信息
    serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);
    ....
}

在創(chuàng)建ServiceInfoHolder時(shí)會(huì)做如下事情

初始化本地緩存目錄
根據(jù)配置從本地緩存初始化服務(wù)，默認(rèn)false
創(chuàng)建FailoverReactor并相互持有ServiceInfoHolder

// 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);
}

本地緩存目錄

在processServiceInfo中會(huì)進(jìn)行本地緩存寫(xiě)入，其實(shí)就是寫(xiě)入這個(gè)目錄(這個(gè)目錄是在創(chuàng)建ServiceInfoHolder時(shí)根據(jù)配置初始化的)，所以目錄中的數(shù)據(jù)正常是最新讀取到的服務(wù)信息

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

本地緩存目錄默認(rèn)路徑：${user.home}/nacos/naming/public，也可以自定義，通過(guò)System.setProperty("JM.SNAPSHOT.PATH")自定義

2. FailoverReactor

在ServiceInfoHolder的構(gòu)造方法中，還會(huì)初始化一個(gè)FailoverReactor類(lèi)，同樣是ServiceInfoHolder的成員變量。FailoverReactor的作用便是用來(lái)處理故障轉(zhuǎn)移的。

構(gòu)造故障轉(zhuǎn)移做了如下事情：

// FailoverReactor
public FailoverReactor(ServiceInfoHolder serviceInfoHolder, String cacheDir) {
    // 持有ServiceInfoHolder的引用
    this.serviceInfoHolder = serviceInfoHolder;
    // 拼接故障目錄：${user.home}/nacos/naming/public/failover
    this.failoverDir = cacheDir + FAILOVER_DIR;
    // 初始化executorService，支持延時(shí)執(zhí)行
    this.executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            // 守護(hù)線程模式運(yùn)行
            thread.setDaemon(true);
            thread.setName("com.alibaba.nacos.naming.failover");
            return thread;
        }
    });
    // 其他初始化操作，通過(guò)executorService開(kāi)啟多個(gè)定時(shí)任務(wù)執(zhí)行
    this.init();
}

init方法執(zhí)行

在這個(gè)方法中開(kāi)啟了三個(gè)定時(shí)任務(wù)，這三個(gè)任務(wù)其實(shí)都是FailoverReactor的內(nèi)部類(lèi)

1. 初始化立即執(zhí)行，然后每間隔5秒再執(zhí)行SwitchRefresher

2. 初始化延遲30分鐘執(zhí)行，執(zhí)行間隔24小時(shí)，執(zhí)行任務(wù)DiskFileWriter

3. 初始化立即執(zhí)行，執(zhí)行間隔10秒，執(zhí)行核心操作為DiskFileWriter

// FailoverReactor
public void init() {
    // 初始化立即執(zhí)行，執(zhí)行間隔5秒，執(zhí)行任務(wù)SwitchRefresher
    executorService.scheduleWithFixedDelay(new SwitchRefresher(), 0L, 5000L, TimeUnit.MILLISECONDS);
    // 初始化延遲30分鐘執(zhí)行，執(zhí)行間隔24小時(shí)，執(zhí)行任務(wù)DiskFileWriter
    executorService.scheduleWithFixedDelay(new DiskFileWriter(), 30, DAY_PERIOD_MINUTES, TimeUnit.MINUTES);
    // 10秒后如果故障目錄為空，則執(zhí)行DiskFileWriter任務(wù)強(qiáng)制備份
    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);
}

DiskFileWriter刷盤(pán)任務(wù)

將ServiceInfo寫(xiě)入備份磁盤(pán)

// 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();
            ...
            // 將緩存寫(xiě)入磁盤(pán)
            DiskCache.write(serviceInfo, failoverDir);
        }
    }
}

SwitchRefresher根據(jù)標(biāo)記故障轉(zhuǎn)移文件切換內(nèi)存標(biāo)志

如果故障轉(zhuǎn)移文件不存在，則直接返回（文件開(kāi)關(guān)）
比較文件修改時(shí)間，如果已經(jīng)修改，則獲取故障轉(zhuǎn)移文件中的內(nèi)容。
故障轉(zhuǎn)移文件中存儲(chǔ)了0和1標(biāo)識(shí)。0表示關(guān)閉，1表示開(kāi)啟。
當(dāng)為開(kāi)啟狀態(tài)時(shí)，執(zhí)行線程FailoverFileReader。

// 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;
                // 獲取故障轉(zhuǎn)移文件內(nèi)容
                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 表示開(kāi)啟故障轉(zhuǎn)移模式
                        if (IS_FAILOVER_MODE.equals(line1)) {
                            switchParams.put(FAILOVER_MODE_PARAM, Boolean.TRUE.toString());
                            new FailoverFileReader().run();
                        // 0 表示關(guān)閉故障轉(zhuǎn)移模式
                        } 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);
        }
    }
}

FailoverFileReader故障后讀取備份文件

該任務(wù)是故障轉(zhuǎn)移的核心，故障轉(zhuǎn)移文件讀取，基本操作就是讀取failover目錄存儲(chǔ)的**備份服務(wù)信息文件**內(nèi)容，然后轉(zhuǎn)換成ServiceInfo，并且將所有的ServiceInfo儲(chǔ)存在FailoverReactor的ServiceMap屬性中。

流程如下：

1. 讀取failover目錄下的所有文件，進(jìn)行遍歷處理

2. 如果文件不存在跳過(guò)

3. 如果文件是故障轉(zhuǎn)移開(kāi)關(guān)標(biāo)志文件跳過(guò)

4. 讀取文件中的備份內(nèi)容，轉(zhuǎn)換為ServiceInfo對(duì)象

5. 將ServiceInfo對(duì)象放入到domMap中

6. 最后判斷domMap不為空，賦值給serviceMap

// 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) {
            // 如果是故障轉(zhuǎn)移標(biāo)志文件，則跳過(guò)
            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;
        }
    }
}·

FailoverReactor.serviceMap的使用

我們?cè)赟erviceInfoHolder中的getServiceInfo方法就會(huì)判斷，如果當(dāng)前是故障切換狀態(tài)，就會(huì)從FailoverReactor中獲取服務(wù)，那么這是就用到了FailoverReactor.serviceMap緩存的服務(wù)了

// 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;
}