分布式开源调度框架TBSchedule详解

雨点打透心脏的1/2处 2022-03-18 10:49 590阅读 0赞

#### **主要内容：** ####

##### **第一部分 TBSchedule基本概念及原理**    #####

##### 1. 概念介绍    #####

##### 2. 工作原理    #####

##### 3. 源码分析    #####

##### 4. 与其他开源调度框架对比 #####

##### **第二部分 TBSchedule分布式调度示例**    #####

##### 1. TBSchedule源码下载    #####

##### 2. 引入源码Demo开发示例    #####

##### 3. 控制台配置任务调度    #####

##### 4. selectTasks方法参数说明    #####

##### 5. 创建调度策略参数说明    #####

##### 6. 创建任务参数说明 #####

--------------------

#### 第一部分 TBSchedule基本概念及原理 ####

##### 1. 概念介绍 #####

TBSchedule是一个支持分布式的调度框架，能让一种批量任务或者不断变化的任务，被动态的分配到多个主机的JVM中，不同的线程组中并行执行。基于ZooKeeper的纯Java实现，由Alibaba开源。

##### 2. 工作原理 #####

TBSchesule对分布式的支持包括[调度机][Link 1]的分布式和执行机的分布式，其网络部署架构图如下：

![这里写图片描述][20161123131752633]

###### 2.1 数据存储 ######

执行机和调度机均以ZooKeeper为注册中心，所有数据以节点及节点内容的形式注册，通过定时汇报主机状态保持存活在ZooKeeper上。

首先看下执行机对ZooKeeper的使用配置文件：

<?xml version="1.0" encoding="utf-8"?>
    <!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">
    <beans default-autowire="byName">
        <bean id="scheduleManagerFactory" class="com.taobao.pamirs.schedule.strategy.TBScheduleManagerFactory" init-method="init">
            <property name="zkConfig">
                <map>
                    <entry key="zkConnectString" value="${schedule.zookeeper.address}" />
                    <entry key="rootPath" value="${schedule.root.catalog}" />
                    <entry key="zkSessionTimeout" value="${schedule.timeout}" />
                    <entry key="userName" value="${schedule.username}" />
                    <entry key="password" value="${schedule.password}" />
                    <entry key="isCheckParentPath" value="true" />
                </map>
            </property>
        </bean>
    </beans>

1）执行机部署启动，会在ZooKeeper上创建永久根节点`schedule.zookeeper.address`，其后所有的操作均在该根节点下进行。

这里以`/ttest/creditjob`为根节点，查看执行机注册后情况：

[zk: 172.26.50.86:2181(CONNECTED) 28] ls /ttest/creditjob
    [strategy, baseTaskType, factory]
    [zk: 172.26.50.86:2181(CONNECTED) 29] ls /ttest/creditjob/factory
    [127.0.0.1$MIE-ZHANGTAO-D1$9D3029EC0C574403B6CFD0C146644A77$0000000000, 127.0.0.1$MIE-ZHANGTAO-D1$D826BC6565DC4D6CB85F7AE321EE51AE$0000000001]

可以看到根节点下面有3个永久子节点，`strategy`存储调度机创建的策略信息，`baseTaskType`存储调度机创建的任务信息，`factory`存储执行机注册的主机信息。每台执行机启动后，都会在`factory`下创建一个`临时顺序子节点`，该节点名是由TBSchedule源码生成的主机唯一表示。

根节点内容为当前TBSchedule内置版本号，可在程序修改，实际没什么意义。

[zk: 172.26.50.86:2181(CONNECTED) 17] get /ttest/creditjob
    tbschedule-3.2.12

2）调度机部署启动，这时不会对ZooKeeper节点做任何操作。打开调度机配置面板：

![这里写图片描述][20161123131950743]

配置好ZooKeeper接入点，点击管理主页，进入调度任务管理面板：

![这里写图片描述][20161123132050760]

输入各项参数创建新任务后，此时会在`baseTaskType`下面创建任务名称永久子节点（调度机所有都宕机重启后，仍能保持数据的完整性），而当前节点的内容就是配置的各项参数。

[zk: 172.26.50.86:2181(CONNECTED) 37] ls /ttest/creditjob/baseTaskType
    [IScheduleTaskDealSingleTest]
    
    [zk: 172.26.50.86:2181(CONNECTED) 39] get /ttest/creditjob/baseTaskType/IScheduleTaskDealSingleTest
    {
       "baseTaskType":"IScheduleTaskDealSingleTest","heartBeatRate":5000,"judgeDeadInterval":60000,"sleepTimeNoData":500,"sleepTimeInterval":0,"fetchDataNumber":500,"executeNumber":1,"threadNumber":1,"processorType":"SLEEP","permitRunStartTime":"0 * * * * ?","expireOwnSignInterval":1.0,"dealBeanName":"iScheduleTaskDealSingleTest","taskParameter":"0","taskKind":"static","taskItems":["0"],"maxTaskItemsOfOneThreadGroup":0,"version":0,"sts":"resume"}

3）创建调度策略，控制调度机调度状态。

![这里写图片描述][20161123132111400]

创建完成调度策略后开启调度，此过程会在对应的任务节点`strategy`下创建`永久子节点`并写入策略数据，在该子节点下创建表示调度机的`临时顺序子节点`并写入调度策略数据。

[zk: 172.26.50.86:2181(CONNECTED) 56] get /ttest/creditjob/strategy/IScheduleTaskDealSingleTest
    {
       "strategyName":"IScheduleTaskDealSingleTest","IPList":["127.0.0.1"],"numOfSingleServer":0,"assignNum":1,"kind":"Schedule","taskName":"IScheduleTaskDealSingleTest","taskParameter":"0","sts":"resume"}
    
    [zk: 172.26.50.86:2181(CONNECTED) 57] get /ttest/creditjob/strategy/IScheduleTaskDealSingleTest/127.0.0.1$MIE-ZHANGTAO-D1$9D3029EC0C574403B6CFD0C146644A77$0000000000
    {
       "strategyName":"IScheduleTaskDealSingleTest","uuid":"127.0.0.1$MIE-ZHANGTAO-D1$9D3029EC0C574403B6CFD0C146644A77$0000000000","requestNum":1,"currentNum":0,"message":""}

同时会在`baseTaskType/IScheduleTaskDealSingleTest`下创建下创建两层`永久子节点`并注册调度主机数据。

[zk: 172.26.50.86:2181(CONNECTED) 45] ls /ttest/creditjob/baseTaskType/IScheduleTaskDealSingleTest/IScheduleTaskDealSingleTest
    [taskItem, server]
    
    [zk: 172.26.50.86:2181(CONNECTED) 50] get /ttest/creditjob/baseTaskType/IScheduleTaskDealSingleTest/IScheduleTaskDealSingleTest/taskItem
    IScheduleTaskDealSingleTest$127.0.0.1$4E8008EE18334564BE1E31C7C0D55296$0000000000,IScheduleTaskDealSingleTest$127.0.0.1$4E8008EE18334564BE1E31C7C0D55296$0000000001
    
    [zk: 172.26.50.86:2181(CONNECTED) 51] get /ttest/creditjob/baseTaskType/IScheduleTaskDealSingleTest/IScheduleTaskDealSingleTest/server  
    reload=true

###### 2.2 分布式高可用高效率保障 ######

1）`调度机的高可用有保障`，多调度机向注册中心注册后，共享调度任务，且同一调度任务仅由一台调度机执行调度，当前调度机异常宕机后，其余的调度机会接上。

2）`执行机的高可用有保障`，多执行机向注册中心注册后，配置执行机单线程（多机总线程为1）执行任务，调度机会随机启动一台执行机执行，当前执行异常机宕机后，调度机会会新调度一台执行机。

3）`执行机的并行高效保障`，配置执行机多线程且划分多任务子项后，各任务子项均衡分配到所有执行机，各执行机均执行，多线程数据一致性协调由任务项参数区分。

4）`弹性扩展失效转移保障`，运行中的执行机宕机，或新增执行机，调度机将在下次任务执行前重新分配任务项，不影响正常执行机任务（崩溃的执行机当前任务处理失效）；运行中的调度机宕机或动态新增调度机，不影响执行机当前任务，调度机宕机后动态切换。

##### 3. 源码分析 #####

###### 3.1 执行机注册节选 ######

从Spring配置文件可以看到，执行机注册的入口在`TBScheduleManagerFactory`的`init`方法，代码片段：

public class TBScheduleManagerFactory implements ApplicationContextAware { 
        public void init() throws Exception {
            Properties properties = new Properties();
            for(Map.Entry<String,String> e: this.zkConfig.entrySet()){
                properties.put(e.getKey(),e.getValue());
            }
            this.init(properties);
        }
    
        public void init(Properties p) throws Exception {
            if(this.initialThread != null){
                this.initialThread.stopThread();
            }
            this.lock.lock();
            try{
                this.scheduleDataManager = null;
                this.scheduleStrategyManager = null;
                ConsoleManager.setScheduleManagerFactory(this);
                if(this.zkManager != null){
                    this.zkManager.close();
                }
                this.zkManager = new ZKManager(p);
                this.errorMessage = "Zookeeper connecting ......" + this.zkManager.getConnectStr();
                initialThread = new InitialThread(this);
                initialThread.setName("TBScheduleManagerFactory-initialThread");
                initialThread.start();
            }finally{
                this.lock.unlock();
            }
        }
    }

`init`方法将配置参数封装到`Properties`对象后开始初始化，连接上ZooKeeper并启动一个新的线程进行节点数据处理。

this.zkManager = new ZKManager(p);
    ...
    initialThread = new InitialThread(this);
    initialThread.start();

跟踪代码可以看到新线程调用的实际处理方法是：

public void initialData() throws Exception{
    
        /** 递归创建永久根节点（/ttest/creditjob）并写入版本信息 */
        this.zkManager.initial();
    
        /** 创建永久子节点 baseTaskType */
        this.scheduleDataManager = new ScheduleDataManager4ZK(this.zkManager);
    
        /** 创建永久子节点 strategy 和 factory */
        this.scheduleStrategyManager  = new ScheduleStrategyDataManager4ZK(this.zkManager);
    
        if (this.start == true) {
    
            /** 注册调度管理器，创建临时顺序子节点，节点表示主机的注册信息 */
            this.scheduleStrategyManager.registerManagerFactory(this);
            if(timer == null){
                timer = new Timer("TBScheduleManagerFactory-Timer");
            }
    
            if(timerTask == null){
                /** 启动一个定时器检测ZooKeeper状态，如果连接失败，关闭所有的任务后，重新连接Zookeeper服务器 */
                timerTask = new ManagerFactoryTimerTask(this);
                timer.schedule(timerTask, 2000,this.timerInterval);
            }
        }
    }

上述几个节点创建完成，并向ZooKeeper注册监听，当有数据变化时获得通知（任务执行/暂停）。到这里，就完成了执行机到ZooKeeper的注册监听过程。

###### 3.2 调度任务创建节选 ######

任务创建提交保存为入口，将参数封装到`ScheduleTaskType`对象中，调用节点创建和更新方法：

//taskTypeEdit.jsp->taskTypeDeal.jsp
    
    if(action.equalsIgnoreCase("createTaskType")){
        ConsoleManager.getScheduleDataManager().createBaseTaskType(taskType);
        result = "任务" + baseTaskType + "创建成功！！！！";
    }else{
        ConsoleManager.getScheduleDataManager().updateBaseTaskType(taskType);
        result = "任务" + baseTaskType + "修改成功！！！！"; 
    }

真正执行任务节点及数据处理的代码段：

//ScheduleDataManager4ZK.java
    
    public void createBaseTaskType(ScheduleTaskType baseTaskType) throws Exception {
        if(baseTaskType.getBaseTaskType().indexOf("$") > 0){
            throw new Exception("调度任务" + baseTaskType.getBaseTaskType() +"名称不能包括特殊字符 $");
        }
    
        /** 在 baseTaskType 节点下创建任务永久节点并写入节点内容为任务配置参数 */
        String zkPath = this.PATH_BaseTaskType + "/"+ baseTaskType.getBaseTaskType();
        String valueString = this.gson.toJson(baseTaskType);
        if ( this.getZooKeeper().exists(zkPath, false) == null) {
            this.getZooKeeper().create(zkPath, valueString.getBytes(), this.zkManager.getAcl(),CreateMode.PERSISTENT);
        } else {
        throw new Exception("调度任务" + baseTaskType.getBaseTaskType() + "已经存在,如果确认需要重建，请先调用deleteTaskType(String baseTaskType)删除");
        }
    }

如果是更新的话，就不会再创建任务永久节点了，直接修改任务节点内容即可。

###### 3.3 策略创建节选 ######

策略创建提交保存为入口，将参数封装到`ScheduleStrategy`对象中，调用节点创建和更新方法：

//scheduleStrategyEdit.jsp->scheduleStrategyDeal.jsp
    
    if (action.equalsIgnoreCase("createScheduleStrategy")) {
        ConsoleManager.getScheduleStrategyManager().createScheduleStrategy(scheduleStrategy);
        isRefreshParent = true;
    } else if (action.equalsIgnoreCase("editScheduleStrategy")) {
        ConsoleManager.getScheduleStrategyManager().updateScheduleStrategy(scheduleStrategy);
        isRefreshParent = true;
    }

真正执行任务节点及数据处理的代码段：

//ScheduleStrategyDataManager4ZK.java
    
    public void createScheduleStrategy(ScheduleStrategy scheduleStrategy) throws Exception {
        String zkPath = this.PATH_Strategy + "/"+ scheduleStrategy.getStrategyName();
    
        /** 在 strategy 节点下创建任务永久节点并写入节点内容为任务配置参数 */
        String valueString = this.gson.toJson(scheduleStrategy);
        if ( this.getZooKeeper().exists(zkPath, false) == null) {
            this.getZooKeeper().create(zkPath, valueString.getBytes(), this.zkManager.getAcl(),CreateMode.PERSISTENT);
        } else {
            throw new Exception("调度策略" + scheduleStrategy.getStrategyName() + "已经存在,如果确认需要重建，请先调用deleteMachineStrategy(String taskType)删除");
        }
    }

如果是更新的话，就不会再创建任务永久节点了，直接修改任务节点内容即可。

###### 3.4 调度控制节选 ######

策略调度控制代码段：

//scheduleStrategyList.jsp->scheduleStrategyDeal.jsp
    
    else if (action.equalsIgnoreCase("deleteScheduleStrategy")) {
        ConsoleManager.getScheduleStrategyManager()
                        .deleteMachineStrategy(
        scheduleStrategy.getStrategyName());
        isRefreshParent = true;
    } else if (action.equalsIgnoreCase("pauseTaskType")) {
        ConsoleManager.getScheduleStrategyManager().pause(
                        scheduleStrategy.getStrategyName());
        isRefreshParent = true;
    } else if (action.equalsIgnoreCase("resumeTaskType")) {
        ConsoleManager.getScheduleStrategyManager().resume(
                        scheduleStrategy.getStrategyName());
        isRefreshParent = true;
    }

真正执行任务节点及数据处理的代码段：

//ScheduleStrategyDataManager4ZK.java
    
    /** 策略删除，即删除strategy下对应的策略节点及数据 */
    public void deleteMachineStrategy(String taskType) throws Exception {
        deleteMachineStrategy(taskType,false);
    }
    
    /** 调度暂停，即修改strategy下对应的策略节点的状态标示数据 */
    public void pause(String strategyName) throws Exception{
        ScheduleStrategy strategy = this.loadStrategy(strategyName);
        strategy.setSts(ScheduleStrategy.STS_PAUSE);
        this.updateScheduleStrategy(strategy);
    }
    
    /** 调度启动，即修改strategy下对应的策略节点的状态标示数据 */
    public void resume(String strategyName) throws Exception{
        ScheduleStrategy strategy = this.loadStrategy(strategyName);
        strategy.setSts(ScheduleStrategy.STS_RESUME);
        this.updateScheduleStrategy(strategy);      
    }

修改节点数据，通过ZooKeeper的事件通知机制，让执行机获得变更通知。

##### 4. 与其他开源调度框架对比 #####

1）`Quartz`：Java事实上的定时任务标准。但Quartz关注点在于定时任务而非数据，并无一套根据数据处理而定制化的流程。虽然Quartz可以基于数据库实现作业的高可用，缺少分布式并行执行作业的功能。

2）`Crontab`：[Linux系统][Linux]级的定时任务执行器，缺乏分布式和集中管理功能。

3）`elastic-job`：当当网最近开源项目，功能跟TBSchedule几乎一样（批斗TBSchedule文档缺失严重），一台服务器只能开启一个任务实例，基于Ip不基于IpPort，单机难调试集群功能。

4）`TBSchedule`：淘宝早期开源，稳定性可以保证。

--------------------

#### 第二部分 TBSchedule分布式调度示例 ####

##### 1. TBSchedule源码下载 #####

下载[TBSchedule源码工程][TBSchedule]，内容包括两部分：工程编译成`jar`的任务开发依赖包和工程编译成`war`的调度控制台。

##### 2. 引入源码Demo开发示例 #####

当前示例与Spring集成，源码可作为普通工程依赖入任务工程，也可将其打包成`jar`并引入依赖，此处版本为`3.2.2.2`。

> `补充`：若打包失败，请检查编译插件版本及jdk编译版本。

任务工程依赖

<dependency>
        <groupId>com.taobao.pamirs.schedule</groupId>
        <artifactId>tbschedule</artifactId>
        <version>3.3.3.2</version>
    </dependency>

调度任务实现`IScheduleTaskDealSingle`，并实现`selectTasks`，`execute`方法，详细示例：

Component("iScheduleTaskDealSingleTest")
    public class IScheduleTaskDealSingleTest implements IScheduleTaskDealSingle<TaskModel> { 
    
        private static final Logger LOG = LoggerFactory.getLogger(IScheduleTaskDealSingleTest.class);
    
        @Override
        public Comparator<TaskModel> getComparator() {
            return null;
        }
    
        @Override
        public List<TaskModel> selectTasks(String taskParameter, String ownSign, int taskQueueNum,
                List<TaskItemDefine> taskItemList, int eachFetchDataNum) throws Exception {
    
            LOG.info("IScheduleTaskDealSingleTest配置的参数，taskParameter:{}，ownSina:{}，taskQueueNum:{},taskItemList:{}, eachFetchDataNum:{}", taskParameter, ownSign, taskQueueNum, taskItemList, eachFetchDataNum);
    
            LOG.info("IScheduleTaskDealSingleTest选择任务列表开始啦..........");
            List<TaskModel> models = new ArrayList<TaskModel>();
            models.add(new TaskModel(String.valueOf(System.currentTimeMillis()), "taosirTest1"));
            models.add(new TaskModel(String.valueOf(System.currentTimeMillis()), "taosirTest2"));
    
            return models;
    
        }
    
        @Override
        public boolean execute(TaskModel model, String ownSign) throws Exception {
    
            LOG.info("IScheduleTaskDealSingleTest执行开始啦.........." + new Date());
            System.out.println(model);
            return true;
    
        }
    
    }

其中`selectTasks`方法获取需要处理的列表（用集合装着），循环集合中的元素并调用`execute`方法执行。子计时任务启动，会直到获取不到数据后才停止等待下一个子计时开始，参数后面详细介绍。

将调度任务注册到`zookeeper`中心，`spring`中引入如下配置：

环境属性配置文件增加如下配置：

#注册中心地址
    schedule.zookeeper.address=172.26.50.86:2181
    #定时任务根目录，任意指定，调度控制台配置时对应
    schedule.root.catalog=/tbschedule/example
    #账户，任意指定，调度控制台配置时对应
    schedule.username=username
    #密码，任意指定，调度控制台配置时对应
    schedule.password=password
    #超时配置
    schedule.timeout=60000

启动容器，`iScheduleTaskDealSingleTest`就完成了到`zookeeper`中心的注册。

> `补充`：TBSchedule提供了IScheduleTaskDealSingle和IScheduleTaskDealMulti两个接口，个人在测试中发现两者除了execute方法上参数不同外，功能上并没有别的不同，只是语义上的区分，在处理模式为SLEEP下getComparator()没有用，一般情况下，都是SLEEP模式。

##### 3. 控制台配置任务调度 #####

将控制台`ScheduleConsole.war`部署到`tomcat`容器。

> `补充`：我通过`ant`运行源码中的`build.xml`构建控制台，部署运行失败（没用过ant，目前不知道原因），这种情况下：

使用方式一，直接用下载包中的控制台部署即可。  
使用方式二，修改工程配置打成`war`包，这灵活，还可以自定义修改，源文件不支持中文，可将编码改成`utf-8`支持。

向注册中心注册配置（跟任务注册用同一根目录，官方wiki图示）  
http://\{server\}:\{port\}/ScheduleConsole/schedule/config.jsp

配置调度任务（官方wiki图示）  
http://\{server\}:\{port\}/ScheduleConsole/schedule/index.jsp

##### 4. selectTasks方法参数说明 #####

`taskParameter`：对应控制台`自定义参数`，可自定义传入做逻辑上的操作

`taskQueueNum`：对应控制台`任务项`数量

`taskItemList`：集合中`TaskItemDefine`的id值对应`任务项`值，多线程处理时，根据任务项协调数据一致性和完整性

`eachFetchDataNum`：对应控制台`每次获取数量`，由于子计时单元开始后，会不断的去取数据进行处理，直到取不到数据子计时才停止，等待下一个子计时开始。可以限制每次取数，防止一次性数据记录过大，内存不足。

`ownSign`：环境参数，一般没什么用

##### 5. 创建调度策略参数说明 #####

`策略名称`：策略标示，可任意填写

`任务类型`：一般保持默认Schedule

`任务名称`：对应任务栏被调度任务名称

`任务参数`：一般不用，保持默认

`单JVM最大线程组数量`：单个JVM允许开启的线程数

`最大线程组数量`：多处理机情况下的线程总数限制（总线程为2，任务项线程为4是没有意义的）

`IP地址`：`127.0.0.1`或者`localhost`会在所有机器上运行，注意多处理机若没有根据任务子项划分数据处理，会导致多处理机重复处理数据，谨慎配置

> 创建示例，官方wiki上有图示，上面主要是各参数的具体含义。

##### 6. 创建任务参数说明 #####

`任务名称`：策略调度的标示，一旦创建保存，不可更改

`任务处理的SpringBean`：注册到spring的任务bean，如`iScheduleTaskDealSingleTest`

`心跳频率`/`假定服务死亡时间`/`处理模式`/`没有数据时休眠时长`/`执行结束时间`：一般保持默认即可

`线程数`：处理该任务的线程数，在没有划分多`任务项`的情况下，多线程是没有意义的，且线程数量大于`任务项`也是没有意义的（线程数小于等于任务项），注意如果开启多线程，必须对数据做任务项过滤

`单线程组最大任务项`：配置单JVM处理的最大任务项数量，多`任务项`情况下，可按需限制，一般默认，多执行机会均衡分配

`每次获取数量`：子计时单元开始，线程会不断的去获取数据（每次获取的限制）并处理数据，直到获取不到数据子计时才结束（方法内不用就可以随意配置）

`每次执行数量`：//还没测试过（可能是将获取的数量拆分多次执行）

`每次处理完休眠时间`：子计时单元开始，只要有数据，就会不停的获取不停的处理，这个时间设置后，子计时单元开始每次获取执行后，不管还有没有待数据，都先[歇会儿][Link 2]再获取处理

`自定义参数`：可自定义控制任务逻辑操作

`任务项`：这项很重要，在多线程情况下，划分任务项是有意义的，但是要注意必须通过任务项参数，协调待处理数据，否则多线程会重复处理

转载地址： [ https://blog.csdn.net/taosir\_zhang/article/details/50728362][https_blog.csdn.net_taosir_zhang_article_details_50728362]

[Link 1]: https://www.baidu.com/s?wd=%E8%B0%83%E5%BA%A6%E6%9C%BA&tn=24004469_oem_dg&rsv_dl=gh_pl_sl_csd
[20161123131752633]: /images/20220318/6da017ac2ba2463cb2b5d7f6ff6615b6.png
[20161123131950743]: /images/20220318/8e1bc0e4f2a34b96ad39fbef0e3c44b0.png
[20161123132050760]: /images/20220318/1b4f379ce92049e0b6d9ad81b9a5c9de.png
[20161123132111400]: /images/20220318/41a884cf765c4f9a93d7a9f9d1acedf2.png
[Linux]: https://www.baidu.com/s?wd=Linux%E7%B3%BB%E7%BB%9F&tn=24004469_oem_dg&rsv_dl=gh_pl_sl_csd
[TBSchedule]: http://code.taobao.org/p/tbschedule/wiki/index
[Link 2]: https://www.baidu.com/s?wd=%E6%AD%87%E4%BC%9A%E5%84%BF&tn=24004469_oem_dg&rsv_dl=gh_pl_sl_csd
[https_blog.csdn.net_taosir_zhang_article_details_50728362]: https://blog.csdn.net/taosir_zhang/article/details/50728362