你好，我是高楼。

这节课，我们继续来讲分布式压测平台的改造。在上一讲，我们已经知道了对象存储和性能监控的实现步骤，这节课，我们来看看分布式调度的改造。

分布式调度平台的选型

在 Java 技术栈中，传统的任务调度实现方案，比如 Quartz、Timer，都存在一些问题，例如：

• 仅单点，不支持集群任务分发；
• 无法统计任务执行数据；
• 没有监控任务执行状态；
• 没有任务异常告警；
• 没有 web 后台管理。

但在如今的全链路压测场景中，大部分的压测平台都是分布式架构，它对任务调度的需求也更高，例如：

• 当同一个压力机或多个压力机的下发任务互斥时，需要做到统一的调度；
• 下发的任务需要支持高可用、实时监控、异常告警等功能；
• 压力机集群需要支持动态、协同进行压测，能应付高并发海量流量的场景；
• 需要统一的管理并能够追踪各压力机任务执行结果，需要快速记录各种任务的属性数据。

也就是说，传统的任务调度实现方案已经不能满足全链路压测平台的需求了。这时候，我们就需要一个分布式任务调度平台。目前业界比较主流的中间件有哪些呢？

目前业界比较主流的中间件主要有Elastic-Job、XXL-JOB和Quartz三种。由于 XXL-JOB 的热度足够（目前关于 XXL-JOB 的 GitHub Stars 有 20.1k 个，登记在用的公司有 446 个），另外它的开发迅速、学习简单、轻量级、易扩展，这些特性都挺适合我们这个全链路压测项目的。综合考虑后，最后我们选择了XXL-JOB。

XXL-JOB 基本概念

既然选定了 XXL-JOB，就要先了解一下 XXL-JOB 的基本结构。

下面是一张 XXL-JOB （V2.1.0）的官网架构图。

从图中我们也可以很直观地看到，XXL-JOB 主要包括两大模块：

• 调度模块（调度中心）：负责管理调度信息，按照调度配置发出调度请求，它自身不承担业务代码。这种调度系统与任务解耦的方式，提高了系统可用性和稳定性，也让调度系统性能不再受限于任务模块。另外，调度中心支持可视化、简单且动态地管理调度信息，例如任务新建、更新、删除、GLUE 开发和任务报警等，所有上述操作都会实时生效。调度中心还支持监控调度结果以及执行日志，支持执行器 Failover。
• 执行模块（执行器）：负责接收调度请求并执行任务逻辑。它可以让任务模块专注于任务的执行等操作，让开发和维护更加简单高效。

为了方便理解，我这里画了一个更加精简的逻辑图：

可以看到，调度器和执行器是独立的。调度器决定任务的调度，并且通过 HTTP 的方式调用执行器接口，执行相应任务。

分布式调度平台的落地思路

那么，我们要怎么通过分布式调度平台来满足我们的需求呢？要清楚这一点，我们就要先理清执行器在流量平台的主要用途，它可以：

1. 管理 GoReplay 的生命周期 ，可以调起和终止 GoReplay 程序；
2. 回传状态（在 GoReplay 启动、结束、或其他标志性事件结束后都会向调度服务回传状态）；
3. 对录制和回放的相关文件进行下载、处理和回传；
4. 实时 Rolling 日志，记录 GoReplay 输出的状态数据，便于实时监控。

对于我们分布式流量平台来说，主要的需求就是流量录制和流量回放操作的调度。我们先来熟悉一下这两种调度方式的落地思路，再来进行具体的实战演示。

流量录制主要过程：

1. 用户设置录制参数，定义录制命令，创建录制任务；
2. 调度中心生成录制任务，并下发到对应网关的执行器上；
3. 执行器收到录制命令，回传状态给调度中心，随即执行对应录制命令，拉起 GoReplay 程序；
4. 录制结束后，GoReplay 程序退出，执行器回传状态给调度中心；
5. 调度中心判定录制任务结束后，下发上传文件子任务到相关执行器上，执行器执行上传命令将对应的流量文件通过 HTTP 上传到对象存储中；
6. 流量录制任务结束。

流量回放的主要过程：

1. 用户设置回放参数，定义回放命令，创建回放任务；
2. 调度中心生成回放任务，并下发到对应压力机的执行器上；
3. 执行器收到回放命令，回传状态给调度服务，随即拉起 GoReplay 程序；
4. 回放结束，GoReplay 退出，执行器回传状态给调度中心；
5. 流量回放任务结束。

好了，了解了流量平台的调度使用过程，我们就来一起看看具体怎样去落地。

分布式调度平台的落地实践

首先，我们需要搭建 XXL-JOB 调度平台，具体的搭建和系统初始化步骤你可以参考官方文档。

搭建成功后，登录调度中心，运行界面如下：

调度中心支持集群部署，这可以提升调度系统的容灾和可用性。不过，在调度中心集群部署时我有几点建议：

• DB 配置保持一致；
• 集群机器时钟保持一致（单机集群忽视）；
• 通过 Nginx 为调度中心集群做负载均衡，分配域名（调度中心访问、执行器回调配置、调用 API 服务等操作均通过该域名进行）。

有了调度平台，我们就可以进行具体的实践了：

• 创建执行器

执行器相当于是包工头，它可以是一个单独项目，也可以集成在其它业务项目里面。要注意的是，执行器首先需要注册到调度中心，这样调度中心才能发现执行器。执行器的执行结果，也需要通过回调的方式反馈给调度中心。

在源码的 xxl-job-executor-samples 里，有 2 个执行的 demo 教我们如何去创建执行器。

这里我用 SpringBoot 项目用来举个例子。首先，要单独拷一个项目出来，如果你想在自己的项目里集成的话，同样也是参考这个 demo，只要在项目 pom 里加上 xxl-job-core 的依赖即可。

然后我们需要修改配置文件 application.properties：

# web port
server.port=8089
# no web
#spring.main.web-environment=false

# log config
logging.config=classpath:logback.xml

### 注册到调度中心
### xxl-job admin address list, such as "http://address" or "http://address01,http://address02"
xxl.job.admin.addresses=http://127.0.0.1:8080/xxl-job-admin

### xxl-job, access token
xxl.job.accessToken=

### 集群部署，这两项配置要一致
### xxl-job executor appname
xxl.job.executor.appname=xxl-job-executor-sample
### xxl-job executor registry-address: default use address to registry , otherwise use ip:port if address is null
xxl.job.executor.address=

### xxl-job executor server-info
xxl.job.executor.ip=127.0.0.1
### 如果要模拟执行器集群部署，打包后单击运行多次，为服务设置随机端口，${random.int[10000,19999]}
xxl.job.executor.port=9999
### 日志存放路径
### xxl-job executor log-path
xxl.job.executor.logpath=./data/applogs/xxl-job/jobhandler
### xxl-job executor log-retention-days
xxl.job.executor.logretentiondays=30

接着，我们在 IDEA 启动项目。

最后，我们在调度中心的【执行器管理】模块里新增一个执行器。

刷新之后就可以看到我们注册好的的执行器了。

到这里，我们的执行器就已经成功被调度中心注册和发现了，下面我们需要创建对应的任务。

• 创建流量录制任务

我们在调度中心新建一个调度任务。这里的运行模式我们选择 GLUE 模式(Shell)。

简单说明一下，运行模式主要分为两种，一种是 BEAN，一种是 GLUE：

• BEAN 模式：在项目中写 Java 类，然后在 JobHandler 里填上 @XxlJob 里面的名字，它是在执行器端代码编写的；
• GLUE 模式：支持 Java、Shell、Python、PHP、Nodejs、PowerShell，以源码方式直接维护在调度中心，该任务实际上是一段对应脚本的执行。

我们知道 GoReplay 的所有操作都是通过命令行组织的，所以 GLUE模式的Shell脚本是满足我们的需求的，且能做到集中式管理。

首先创建任务，点击该任务右侧的“GLUE”按钮，前往 GLUE 任务的 Web IDE 界面。这个界面支持对任务代码进行开发（也可以在 IDE 中开发完成后，复制粘贴到 Web IDE 中）。

你可以看看我们在 IDE 中开发的流量录制命令。

#!/bin/bash
echo "xxl-job: hello shell"

echo "脚本位置：$0"
echo "任务参数：$1"
echo "分片序号 = $2"
echo "分片总数 = $3"

cd /Users/GoProjects/src/goreplay/examples/middleware
./goreplay --input-raw :8201 --output-file=request-mall-all.gor -output-file-append --http-set-header "dunshan:7DGroup" --http-set-header "User-Agent:Replayed-by-Gor" --input-raw-track-response --prettify-http --http-disallow-url /actuator/health

echo "Good bye!"
exit 0

刚才我们说过，因为GLUE这个模式的任务实际上是一段 “Shell” 脚本，所以任务建好后会存在一个安全问题：我们没有做鉴权。要解决这个问题方法也很简单，只需要在调度中心和执行器的 application.properties 里加上相同的 token 即可。

任务建好后，在执行时，我们可以传入任务参数，这里我们传入的是监听端口号8081。

好了，开发完成后，我们点击启动：

点击执行日志右侧的 “执行日志” 按钮，可跳转至执行日志界面查看业务代码中打印的完整日志：

点击“日志”按钮，可以查看任务历史调度日志，每次任务调度的调度结果、执行结果等。点击“执行日志”按钮可查看执行器完整日志。

好了，到这里我们的流量录制任务就已经定义完了。至于流量回放、文件上传等其它任务的套路和这个类似。你可以按照差不多的步骤进行操作，遇到问题的话也可以来跟我讨论。

• 使用子任务

当我们需要做流量录制时，首先我们需要执行程序初始化任务，然后再执行流量录制任务，最后，我们需要将产生的流量文件上传到对象存储（流量仓库）中。那么，我们要怎么保证数据在三个任务之间顺利传递呢？

解决这种问题的思路一般有两种：

• 第一种就是写成一个大 job，让任务串行化执行；
• 第二种就是使用多个子任务，在一个上游任务末尾触发另一个子任务。

如果我们想要在本任务成功执行后触发另外一个任务，可以把另外的任务作为本任务的子任务执行，它的操作也很简单，只需要在子任务一栏填上任务 id 即可：

• 压测任务分片

全链路压测中一个显著的特点就是能够实现海量数据的并发请求，但这对于单线程的普通的任务来说是存在瓶颈的。这个时候如果你想提高并发，那么自然就要多线程跑嘛。但是用多线程跑数据也会存在问题，比如我开 3 个执行器，但你会发现只有个别执行器在拼命工作，其它的都处在休息状态。这是一个典型的任务分配不均的问题。

这时候就要用到我们的分片任务了，也就是要多台执行器共同完成任务。

但这里又有一个问题。比如流量回放的时候，多台压力机都执行同一个流量文件，相同的请求三台压力机都发一遍，那岂不是乱套了。

这里的解决思路也很简单，我们让每台执行器处理1/3的流量，大家把需要回放的流量文件平均切分了嘛，我干 1/3，你干 1/3，它干 1/3，这样就不会产生冲突了。

分片任务在运行的时候，调度器会给每个执行器发送一个不同的分片序号，分片的最大序号跟执行器的总数量是一样的，这样可以确保每个执行器都会执行到这个任务。

比如上图中第一个执行器拿到分片序号 0，第二台执行器拿到分片序号 1，第三台执行器拿到分片序号 2。那现在就好办了，我们只需要把处理的流量文件进行模 3 取余，余数为 0 的文件就由执行器 1 跑，余数为 1 的文件就由执行器 2 跑，余数为 2 的文件就由执行器 3 跑。

具体的解决方案，我们可以使用路由策略「分片广播」来调度任务：

1. 通过“任务参数”传入执行任务压力机数量；
2. 在定时任务逻辑中，根据获取到的分片参数、执行任务压力机数量，决策当前节点是否需要执行，切分流量文件并处理：
   • 如果 分片序号 > (执行压力机数量 - 1)，则当前节点不执行任务，直接返回；
   • 否则，取「分片序号」和 「压力机数量」 作为分片参数，流量文件并处理。

这样，我们可以实现灵活调度 [1, N] 个节点并行执行压测任务。

我们切分流量文件的代码可以这样写。

package com.xxl.job.executor.until;

import cn.hutool.core.io.file.FileReader;
import cn.hutool.core.io.file.FileWriter;
import org.junit.jupiter.api.Test;
import org.junit.platform.commons.util.StringUtils;

import java.util.*;

/**
 * @author dunshan
 * @description: 分割流量文件
 * @date 2021-12-04 11:11:11
 */
public class NcountGor {



    @Test
    public void countTxt() {
        //分组
        int n = 3;

        LinkedHashSet<String> total = new LinkedHashSet<>();
        //读取文件
        FileReader fileReader = new FileReader("/Users/2021/10/request-mall-all.gor");
        String result = countTotal(total, fileReader);
        System.out.println("total:" + total.size());
        //放入 list 中进行分组使用
        List<String> list = new ArrayList<>();
        //组装分割文件
        for (String str : total) {
            list.add(str);
        }
        saveFile(n, result, list);

    }

    /**
     * 根据分割数据把文件切割文件
     *
     * @param n      切割数据
     * @param result 文件内容
     * @param list   分组数据
     */
    private void saveFile(int n, String result, List<String> list) {
        int number = (list.size() / n) + 1;
        List<List<String>> lists = groupList(list, number);
        for (int i = 0; i < lists.size(); i++) {
            //保存文件
            String path = String.format("/Users/2021/10/11/request_%s.gor", i);
            FileWriter writer = new FileWriter(path);
            for (int j = 0; j < lists.get(i).size(); j++) {
                //分割数据
                String[] content = result.split("\uD83D\uDC35\uD83D\uDE48\uD83D\uDE49");
                for (String s : content) {
                    if (s.contains(lists.get(i).get(j))) {
                        writer.append(s);
                        //拼接切割文件
                        writer.append("\uD83D\uDC35\uD83D\uDE48\uD83D\uDE49");
                    }
                }
            }
        }
    }

    /**
     * 统计流量文件总数
     *
     * @param total
     * @param fileReader
     * @return
     */
    private String countTotal(LinkedHashSet<String> total, FileReader fileReader) {
        String result = fileReader.readString();
        String[] twoLine = result.split("\r\n");
        for (String str : twoLine) {
            if (StringUtils.isNotBlank(str)) {
                //转换字符数组
                char[] chars = str.toCharArray();
                //判断第一个字母是否是一
                if (String.valueOf(chars[0]).equals("1")) {
                    total.add(str.substring(2, 26));
                }
            }
        }
        return result;
    }

    /**
     * 流量文件分组
     *
     * @param list 流量文件
     * @param n    每个数组多少
     * @return 分组
     */
    private static List<List<String>> groupList(List<String> list, Integer n) {
        // 求余数
        int remainder = list.size() % n;
        // 求分组数
        int count0 = list.size() / n;
        boolean flag = false;
        if (remainder >= n / 1) {
            flag = true;
        }
        List<List<String>> data = new ArrayList<>();
        // 获取多余 list
        List<String> sub = new ArrayList<>(list.subList(list.size() - remainder, list.size()));
        // 遍历 list 到余数前
        for (int i = 0; i < count0 + 1; i++) {
            List<String> ls;
            if (flag) {
                if (i == count0) {
                    break;
                }
                ls = new ArrayList<>(list.subList(i * n, i * n + n));
                // 将余出的数据加入分组
                if (i < sub.size()) {
                    ls.add(sub.get(i));
                }
            } else {
                int endex = i * n + n;
                if (endex >= list.size()) {
                    endex = list.size();
                }
                ls = new ArrayList<>(list.subList(i * n, endex));
            }
            if (ls.size() != 0) {
                data.add(ls);
            }
        }
        return data;
    }

}

切分的文件效果如下：

JobHandler 中的实现可以参考下面这种写法：

	/**
     * 2、分片广播任务
     */
    @XxlJob("shardingJobHandler")
    public void multiMachineMultiTasks() throws Exception {
        String param = XxlJobHelper.getJobParam();
        if (StringUtils.isBlank(param)) {
            XxlJobHelper.log("任务参数为空");
            XxlJobHelper.handleFail();
            return;
        }

        //读取文件
        FileReader fileReader = new FileReader(param);
        //总分片数
        String shardTotal = countTotal(total, fileReader);
        //分片参数
        int shardIndex = XxlJobHelper.getShardIndex();

        XxlJobHelper.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);

        LinkedHashSet<String> total = new LinkedHashSet<>();

        //放入 list 中进行分组使用
        List<String> list = new ArrayList<>();
        //组装分割文件
        for (String str : total) {
            list.add(str);
        }
        saveFile(n, result, list);

        // 业务逻辑
        for (int i = 0; i < shardTotal; i++) {
            if (i == shardIndex) {
                XxlJobHelper.log("第 {} 片, 命中分片开始处理", i);
                //分片流量回放
                process(shardIndex, shardTotal);
            } else {
                XxlJobHelper.log("第 {} 片, 忽略", i);
            }
        }

    }

新建任务的时候，我们选择分片任务，填上对应的 JobHandler 即可。

最后需要说明一下，分片的数据量不一定是完全均等的，上面的取模只是一个例子、一个思路。我们也可以把 0、1、2 替换成其他条件去从所有数据中获取部分数据。比如分片序号是 0 的压力机回放网关 1 的流量文件，分片序号是 1 的压力机回放网关 2 的流量文件，分片序号是 2 的压力机我回放网关 3 的流量文件，具体怎么分我们可以自由选择。

通用 RESTful API 支持

我们知道，线上压测是一项对风险管控要求更高的活动，压测平台需要有更及时的异常预警机制，也就是说，必须要有压测实时监控和异常熔断能力。这就要求我们的调度任务能实时对接第三方的监控系统。

而XXL-JOB 目标就是一种跨平台、跨语言的任务调度规范和协议。针对第三方监控系统，比如Prometheus，我们可以借助 XXL-JOB 的标准 RESTful API 方便地实现任务回调支持。

标准 RESTful API 主要包括两部分：

• 调度中心 RESTful API：调度中心提供给执行器使用的 API；不局限于官方执行器使用，第三方可使用该 API 来实现执行器。主要包括：执行器注册、任务结果回调等API；
• 执行器 RESTful API：执行器提供给调度中心使用的 API；官方执行器默认已实现，第三方执行器需要实现并对接提供给调度中心，任务触发、任务终止、任务日志查询等API。

除此之外，如果我们有需要通过 API 操作调度中心，可以个性化扩展 “调度中心 RESTful API” 并使用。

总结

流量平台是一个全链路压测项目的引擎。但对于一个大企业来说，分布式的流量平台是一个必要的条件。

在全链路分布式流量平台改造中，我强调了几个特性：高可靠、高性能、低成本、高效率。

我们在做技术改造的时候，一定要牢记下面几项关键原则：

• 基于开源，拥抱开源，尽量使用成熟的第三方框架，不重复造轮子；
• 技术栈和组件要适合研发团队技术能力及主流技术方向；
• KISS 原则(Keep It Simple and Stupid)；
• 尽量考虑 ROI（投入产出比）；
• 不断演进迎合公司业务发展。大公司方案不一定是适合的，要避免为技术而技术；
• 任何方案不必追求大而全，可以在实践中不断完善和演进；
• 需求导向，借力开源，科学选型，快速集成，重视扩展，演进改善。

刚才，我还把一个分布式流量平台的需求以及需要改造到什么地步，都给你梳理了一遍。希望能给你一些借鉴。

思考题

学完这节课，请你思考两个问题：

1. 你有没有接触过分布式流量平台的改造工作，请谈谈你的心得。
2. 关于分布式调度框架，你有没有接触过其它的项目？

欢迎你在留言区与我交流讨论。当然了，你也可以把这节课分享给你身边的朋友，他们的一些想法或许会让你有更大的收获。我们下节课见！