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

说明:本文不介绍如何使用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中的。

七、Diamond客户端与服务端交互时序图

图片 1.png

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Java3y

Shiro入门这篇就够了【Shiro的基础知识、回顾URL拦截】

前言 本文主要讲解的知识点有以下: 权限管理的基础知识 模型 粗粒度和细粒度的概念 回顾URL拦截的实现 Shiro的介绍与简单入门 一、Shiro基础知识 在...

95470
来自专栏名山丶深处

springboot集成schedule(深度理解)

79850
来自专栏java一日一条

HDFS知识点总结

HDFS是什么:HDFS即Hadoop分布式文件系统(Hadoop Distributed Filesystem),以流式数据访问模式来存储超大文件,运行于商用...

13120
来自专栏安恒网络空间安全讲武堂

Python编写渗透工具学习笔记二 | 0x02利用FTP与web批量抓肉鸡

0x02利用FTP与web批量抓肉鸡 脚本要实现的目标和思路: 先尝试匿名登录ftp,当匿名登录失败时再尝试用用户/密码爆破登录,登录成功后,脚本会搜索ftp中...

1.8K70
来自专栏数据之美

实战 windows7 下 eclipse 远程调试 linux hadoop

恩,之所以有这篇博客,是因为最近又有童鞋咨询怎么在 windows 7 下用eclipse远程调试部署在linux下的hadoop,其实我自己不这么混搭的,既然...

30880
来自专栏小白鼠

Hadoop集群搭建Linux环境准备基础配置安装HadoopHA集群安装HIVE安装MySQL安装HBASE安装Flume问题总结

本文主要讲解了Hadoop集群环境的搭建过程,实际应用中应该不会这样做,而是通过一些管理工具进行安装,比如可视化安装:Ambari。

56320
来自专栏Albert陈凯

Hadoop数据分析平台实战——180Oozie工作流使用介绍离线数据分析平台实战——180Oozie工作流使用介绍

离线数据分析平台实战——180Oozie工作流使用介绍 Oozie工作流介绍 Oozie的四大组件服务分别是: workflow, coordinator, b...

41750
来自专栏葡萄城控件技术团队

Winform文件下载之WinINet

在C#中,除了webclient我们还可以使用一组WindowsAPI来完成下载任务。这就是Windows Internet,简称 WinINet。本文通过一个...

23480
来自专栏Java帮帮-微信公众号-技术文章全总结

day07.HDFS学习【大数据教程】

分而治之:将大文件、大批量文件,分布式存放在大量服务器上,以便于采取分而治之的方式对海量数据进行运算分析;

26140
来自专栏王磊的博客

entity framework使用技巧

1、无需先查询数据的修改方法 // 将创建的data实体添加到ObjectContext db.Data.Attach(data); // 手动设置状态为修改 ...

27950

扫码关注云+社区

领取腾讯云代金券