深入淘宝Diamond之客户端架构解析

小程故事多

发布于 2018-08-22 10:30:52

2.3K0

发布于 2018-08-22 10:30:52

说明：本文不介绍如何使用Diamond，只介绍Diamond的实现原理

一、什么是Diamond

diamond是淘宝内部使用的一个管理持久配置的系统，它的特点是简单、可靠、易用，目前淘宝内部绝大多数系统的配置，由diamond来进行统一管理。 diamond为应用系统提供了获取配置的服务，应用不仅可以在启动时从diamond获取相关的配置，而且可以在运行中对配置数据的变化进行感知并获取变化后的配置数据。持久配置是指配置数据会持久化到磁盘和数据库中。

二、Diamond的特点

简单：整体结构非常简单，从而减少了出错的可能性。
可靠：应用方在任何情况下都可以启动，在承载淘宝核心系统并正常运行一年多以来，没有出现过任何重大故障。
易用：客户端使用只需要两行代码，暴露的接口都非常简单，易于理解。

三、Diamond的持久机制

Paste_Image.png

订阅方获取配置数据时，直接读取服务端本地磁盘文件，尽量减少对数据库压力。这种架构用短暂的延时换取最大的性能和一致性，一些配置不能接受延时的情况下，通过API可以获取数据库中的最新配置

四、Diamond的容灾机制

Diamond作为一个分布式环境下的持久配置系统，有一套完备的容灾机制，数据被存储在：数据库，服务端磁盘，客户端缓存目录，以及可以手工干预的容灾目录。客户端通过API获取配置数据按照固定的顺序去不同的数据源获取数据：容灾目录，服务端磁盘，客户端缓存。

• 数据库主库不可用，可以切换到备库，Diamond继续提供服务 • 数据库主备库全部不可用，Diamond通过本地缓存可以继续提供读服务 • 数据库主备库全部不可用，Diamond服务端全部不可用，Diamond客户端使用缓存目录继续运行，支持离线启动 • 数据库主备库全部不可用，Diamond服务端全部不可用，Diamond客户端缓存数据被删，可以通过拷贝备份的缓存目录到容灾目录下继续使用

五、Diamond的架构图

图片 1.png

六、Diamond订阅端(客户端)分析

先看一个简单的客户端订阅代码实现：

public class DiamondTestClient {
    public static DiamondManager manager;
    public static void main(String[] str) {
        initDiamondManager();
    }
    private static void initDiamondManager() {
        manager = new DefaultDiamondManager("group_test", "dataId_test", new ManagerListener() {
            public void receiveConfigInfo(String configInfo) {
                System.out.println("configInfo="+ configInfo);
            }   
        });   
    }  
}

参数的说明： DefaultDiamondManager有三个参数分别是：groupId，dataId和listener。 group和dataId为String类型，二者结合为diamond-server端保存数据的惟一key ManagerListener 是客户端注册的数据监听器，它的作用是在运行中接受变化的配置数据，然后回调receiveConfigInfo()方法，执行客户端处理数据的逻辑。如果要在运行中对变化的配置数据进行处理，就一定要注册ManagerListener 我们来看一下DefaultDiamondManager的类图

Paste_Image.png

DefaultDiamondManager的构造方法代码如下：

public DefaultDiamondManager(String group, String dataId, ManagerListener managerListener) {
        this.dataId = dataId;
        this.group = group;

        diamondSubscriber = DiamondClientFactory.getSingletonDiamondSubscriber();

        this.managerListeners.add(managerListener);
        ((DefaultSubscriberListener) diamondSubscriber.getSubscriberListener()).addManagerListeners(this.dataId,
            this.group, this.managerListeners);
        diamondSubscriber.addDataId(this.dataId, this.group);
        diamondSubscriber.start();

    }

说明 1、利用工厂类DiamondClientFactory创建单例订阅者类。 2、将客户端创建的侦听器类添加到侦听器管理list中并注入到新创建的订阅者类中。 3、为订阅者设置dataId和groupId。 4、启动订阅者线程，开始轮询消息。

DiamondSubscriber的类图如下：

Paste_Image.png

执行diamondSubScriber.start()方法直接进入DefaultDiamondSubscriber子类中，先看如下代码：

/**
     * 启动DiamondSubscriber：<br>
     * 1.阻塞主动获取所有的DataId配置信息<br>
     * 2.启动定时线程定时获取所有的DataId配置信息<br>
     */
    public synchronized void start() {
        if (isRun) {
            return;
        }

        if (null == scheduledExecutor || scheduledExecutor.isTerminated()) {
            scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
        }

        localConfigInfoProcessor.start(this.diamondConfigure.getFilePath() + "/" + DATA_DIR);
        serverAddressProcessor = new ServerAddressProcessor(this.diamondConfigure, this.scheduledExecutor);
        serverAddressProcessor.start();

        this.snapshotConfigInfoProcessor =
                new SnapshotConfigInfoProcessor(this.diamondConfigure.getFilePath() + "/" + SNAPSHOT_DIR);
        // 设置domainNamePos值
        randomDomainNamePos();
        initHttpClient();

        // 初始化完毕
        isRun = true;

        if (log.isInfoEnabled()) {
            log.info("当前使用的域名有：" + this.diamondConfigure.getDomainNameList());
        }

        if (MockServer.isTestMode()) {
            bFirstCheck = false;
        }
        else {
            // 设置轮询间隔时间
            this.diamondConfigure.setPollingIntervalTime(Constants.POLLING_INTERVAL_TIME);
        }
        // 轮询
        rotateCheckConfigInfo();

        addShutdownHook();
    }

说明： 1、ServerAddressProcessor类从服务端获取提供服务的地址列表（可能会多个）。 2、randomDomainNamePos这个方法是随机从服务地址列表中选取一个地址。 3、初始化httpClient客户端，使用initHttpClient方法。 4、设置读取配置文件的轮询时间默认为15秒。 5、rotateCheckConfigInfo这个方法是真正与服务端交互的轮询方法。

rotateCheckConfigInfo方法的代码如下：

/**
     * 循环探测配置信息是否变化，如果变化，则再次向DiamondServer请求获取对应的配置信息
     */
    private void rotateCheckConfigInfo() {
        scheduledExecutor.schedule(new Runnable() {
            public void run() {
                if (!isRun) {
                    log.warn("DiamondSubscriber不在运行状态中，退出查询循环");
                    return;
                }
                try {
                    checkLocalConfigInfo();
                    checkDiamondServerConfigInfo();
                    checkSnapshot();
                }
                catch (Exception e) {
                    e.printStackTrace();
                    log.error("循环探测发生异常", e);
                }
                finally {
                    rotateCheckConfigInfo();
                }
            }

        }, bFirstCheck ? 60 : diamondConfigure.getPollingIntervalTime(), TimeUnit.SECONDS);
        bFirstCheck = false;
    }

说明 1、方法内部启动一个定时线程，默认每隔60秒执行一次。 2、方法内部实际上三个主方法分别是:

checkLocalConfigInfo：主要是检查本地数据是否有更新，如果没有则返回，有则返回最新数据，并通知客户端配置的listener。
checkDiamondServerConfigInfo：远程调用服务端，获取最新修改的配置数据并通知客户端listener。
checkSnapshot：主要是持久化数据信息用的方法。

6.1 checkLocalConfigInfo代码分析

private void checkLocalConfigInfo() {
        for (Entry<String/* dataId */, ConcurrentHashMap<String/* group */, CacheData>> cacheDatasEntry : cache
            .entrySet()) {
            ConcurrentHashMap<String, CacheData> cacheDatas = cacheDatasEntry.getValue();
            if (null == cacheDatas) {
                continue;
            }
            for (Entry<String, CacheData> cacheDataEntry : cacheDatas.entrySet()) {
                final CacheData cacheData = cacheDataEntry.getValue();
                try {
                    String configInfo = getLocalConfigureInfomation(cacheData);
                    if (null != configInfo) {
                        if (log.isInfoEnabled()) {
                            log.info("本地配置信息被读取, dataId:" + cacheData.getDataId() + ", group:" + cacheData.getGroup());
                        }
                        popConfigInfo(cacheData, configInfo);
                        continue;
                    }
                    if (cacheData.isUseLocalConfigInfo()) {
                        continue;
                    }
                }
                catch (Exception e) {
                    log.error("向本地索要配置信息的过程抛异常", e);
                }
            }
        }

说明： 1、循环本地缓存数据，比较数据是否更新变化，重点看getLocalConfigureInfomation方法。 2、如果有更新数据则调用popConfigInfo方法通知客户端listener。

再深入看getLocalConfigureInfomation方法，代码如下：

// 判断是否变更，没有变更，返回null
        if (!filePath.equals(cacheData.getLocalConfigInfoFile())
                || existFiles.get(filePath) != cacheData.getLocalConfigInfoVersion()) {
            String content = FileUtils.getFileContent(filePath);
            cacheData.setLocalConfigInfoFile(filePath);
            cacheData.setLocalConfigInfoVersion(existFiles.get(filePath));
            cacheData.setUseLocalConfigInfo(true);

            if (log.isInfoEnabled()) {
                log.info("本地配置数据发生变化, dataId:" + cacheData.getDataId() + ", group:" + cacheData.getGroup());
            }

            return content;
        }
        else {
            cacheData.setUseLocalConfigInfo(true);

            if (log.isInfoEnabled()) {
                log.debug("本地配置数据没有发生变化, dataId:" + cacheData.getDataId() + ", group:" + cacheData.getGroup());
            }

            return null;
        }

说明： 这段代码很关键，判断当前缓存的数据是否持久化的文件数据是否一致，包括版本号，文件路径等信息，如果服务器端有配置数据更新，客户端则拿到最新的数据后更新本地文件内容。

popConfigInfo方法的代码如下：

void popConfigInfo(final CacheData cacheData, final String configInfo) {
        final ConfigureInfomation configureInfomation = new ConfigureInfomation();
        configureInfomation.setConfigureInfomation(configInfo);
        final String dataId = cacheData.getDataId();
        final String group = cacheData.getGroup();
        configureInfomation.setDataId(dataId);
        configureInfomation.setGroup(group);
        cacheData.incrementFetchCountAndGet();
        if (null != this.subscriberListener.getExecutor()) {
            this.subscriberListener.getExecutor().execute(new Runnable() {
                public void run() {
                    try {
                        subscriberListener.receiveConfigInfo(configureInfomation);
                        saveSnapshot(dataId, group, configInfo);
                    }
                    catch (Throwable t) {
                        log.error("配置信息监听器中有异常，group为：" + group + ", dataId为：" + dataId, t);
                    }
                }
            });
        }
        else {
            try {
                subscriberListener.receiveConfigInfo(configureInfomation);
                saveSnapshot(dataId, group, configInfo);
            }
            catch (Throwable t) {
                log.error("配置信息监听器中有异常，group为：" + group + ", dataId为：" + dataId, t);
            }
        }
    }

说明： 这段代码主要是将已经更新的数据通知给客户端织入的listener程序，使能够达到最新数据通知给客户端。

6.2 checkDiamondServerConfigInfo代码分析

private void checkDiamondServerConfigInfo() {
        Set<String> updateDataIdGroupPairs = checkUpdateDataIds(diamondConfigure.getReceiveWaitTime());
        if (null == updateDataIdGroupPairs || updateDataIdGroupPairs.size() == 0) {
            log.debug("没有被修改的DataID");
            return;
        }
        // 对于每个发生变化的DataID，都请求一次对应的配置信息
        for (String freshDataIdGroupPair : updateDataIdGroupPairs) {
            int middleIndex = freshDataIdGroupPair.indexOf(WORD_SEPARATOR);
            if (middleIndex == -1)
                continue;
            String freshDataId = freshDataIdGroupPair.substring(0, middleIndex);
            String freshGroup = freshDataIdGroupPair.substring(middleIndex + 1);

            ConcurrentHashMap<String, CacheData> cacheDatas = cache.get(freshDataId);
            if (null == cacheDatas) {
                continue;
            }
            CacheData cacheData = cacheDatas.get(freshGroup);
            if (null == cacheData) {
                continue;
            }
            receiveConfigInfo(cacheData);
        }
    }

说明： 1、通过HttpClient方式从服务端获取更新过的dataId和groupId集合。 2、根据dataId和groupId再从服务端将相应变化的数据获取下来。 3、通知客户端注册的listener程序。

上面二种方式通知客户端的listener程序，都是通过allListeners这个属性获取的

private final ConcurrentMap<String/* dataId + group */, CopyOnWriteArrayList<ManagerListener>/* listeners */> allListeners =
            new ConcurrentHashMap<String, CopyOnWriteArrayList<ManagerListener>>();

这行代码就是在最开始的那个客户端使用的例子中注册在allListeners中的。