基于腾讯云API构建任务调度框架

背景

很多开发者会基于云厂商提供的API或者SDK进行二次开发，但是可能因为不熟悉云上资源的特点，或是难以找到API/SDK优雅的使用姿势，导致二次开发的过程中困难重重。笔者在本文中，将为大家介绍一套适用于使用API/SDK控制云资源的分布式任务调度框架，以及对此框架的瓶颈分析和优化思路。这套框架已经在腾讯云多款PAAS产品中经受了考验，是高效而稳定的。

在分布式的任务调度框架中，通常会使用TASK-STEP的结构对任务进行切分，将一个大而复杂的任务TASK，拆解成一个个小而简单的步骤STEP，通过跟踪STEP的完成进度，来判断TASK的整体进展，在这种模式下，框架中的消费者往往承担着很关键的工作，消费者的消费速度直接决定了系统整体的吞吐量。

架构演变

1. TASK 的定义与特点

例如我们将【创建CVM】定义为一个 TASK，对应的 STEP 如下

1. CreateInstance 发起创建 调用 CVM RunInstances
2. VerifyInstance 检查创建结果 调用 CVM DescribeInstances
3. RecordResult   保存记录，更新 DB

我们将 STEP 执行的总次数记为这个 TASK 的 workload，考虑到一些STEP会被执行多次，TASK的 workload >= STEP 数量。

仍以【创建CVM】为例，CVM 发起创建后，假设需要30秒完成创建，那么在CreateInstance步骤之后立刻执行VerifyInstance大概率会出现失败，此时我们每隔几秒重试步骤VerifyInstance，直到30秒后，VerifyInstance步骤成功，我们再进入到第三步RecordResult，假设 TASK 执行期间我们共执行步骤VerifyInstance 10次，CreateInstance和RecordResult各1次，那么这个 TASK 最终的 workload 就是 12。

实际执行效果

1. CreateInstance 发起创建 调用 CVM RunInstances
2. VerifyInstance 检查创建结果 调用 CVM DescribeInstances CVM:PENDING
2. VerifyInstance 检查创建结果 调用 CVM DescribeInstances CVM:PENDING
2. VerifyInstance 检查创建结果 调用 CVM DescribeInstances CVM:PENDING
2. VerifyInstance 检查创建结果 调用 CVM DescribeInstances CVM:PENDING
2. VerifyInstance 检查创建结果 调用 CVM DescribeInstances CVM:PENDING
2. VerifyInstance 检查创建结果 调用 CVM DescribeInstances CVM:PENDING
2. VerifyInstance 检查创建结果 调用 CVM DescribeInstances CVM:PENDING
2. VerifyInstance 检查创建结果 调用 CVM DescribeInstances CVM:PENDING
2. VerifyInstance 检查创建结果 调用 CVM DescribeInstances CVM:PENDING
2. VerifyInstance 检查创建结果 调用 CVM DescribeInstances CVM:RUNNING
3. RecordResult   保存记录，更新 DB

从以上的例子可以看出，对于 TASK 而言，有几个明显的特点：

• TASK workload 未知，不同类型的 TASK workload 不同；相同类型的 TASK，每次执行的 workload 也不一定相同
• TASK IO密集，每一次 STEP 的执行往往代表了一次IO操作，workload 越大，代表IO次数越多

2. 最初架构

整体架构

• API: 负责用户的交互，接收用户的请求，并生成 TASK
• Scheduler：调度器，通过 RabbitMQ 下发 TASK
• Subscriber：TASK 实际执行者，从 RabbitMQ 接收并执行 TASK
• Cronplugin：周期性检查模块，可以灵活插拔调度策略

结合对TASK的分析，我们设计了如上的架构。TASK 的调度采用了 work-sharing（工作共享）方案，Subscriber 公平地从 MQ 中获取最新待执行的 TASK，在 Subscriber 进程内执行 TASK，并将结果返回给 Scheduler。

我们为 TASK 的消费者设计了以下特性：

• 灵活：STEP 的数量、顺序未来均为可能会发生改变，消费者要能灵活地适应这种改变
• 事务性：TASK 如果中途执行失败，要进行必要的回滚，保证不残留脏数据
• 高吞吐：从系统整体的角度来看，我们希望 TASK 能同时处理，不希望出现 RabbitMQ 中 TASK 消息出现堆积
• 低时延：当 TASK 对应的操作结束时，我们希望 TASK 也能立刻结束

围绕这四点，我们开发了框架内的消费者 Subscriber，Subscriber 将 TASK 的 STEP 执行顺序抽象为DAG，每个 STEP 均配置了成功步骤、失败步骤、重试步骤，根据当前步骤的结果，能唯一确定下一个待执行的步骤，TASK 一旦开始，STEP便“奔流到海不复回”，这样的设计同时兼顾了灵活与事务，当加入新的 STEP 时，我们只需要为新 STEP 在 DAG 中找到位置加入即可，当 STEP 出现失败，需要回滚，只需要按照失败步骤往下执行即可。

消费者 Subscriber 的内部架构如下，Subscriber 内部采用了 Reactor 模型，实现了一个高性能的消费者线程池，使得同一个 Subscriber 进程能同时处理多个 TASK，大大提升系统整体的 TASK 吞吐量。

配合内核提供的高性能 eventfd/timerfd + epoll，让事件在进程内部快速而及时的传递，有效降低了 TASK 的平均处理时延。

Subscriber 架构

当 Subscriber 从 RabbitMQ 获取到【CreateCVM】的 TASK后，该 TASK 的处理流程如下：

1. 从 TASK-STEP-CONFIG 读取 TASK 的第一个 STEP
2. 将与 STEP 执行相关的信息(exec_info)加入 Execute Queue
3. MasterThread 从队列中获取 exec_info，判断下一步该执行的 STEP
4. 如果要执行新的 STEP，为新的 STEP 创建事件(eventfd)，并加入事件循环；如果要重试 STEP，为当前 STEP 创建定时事件(timerfd)，加入事件循环
5. WorkerThread 收到事件后，获取该事件对应的 exec_info，然后开始工作
6. WorkerThread 完成工作后，将该 STEP 的结果加入 Execute Queue，再由 MasterThread 判断是该继续下一个 STEP，还是重试当前 STEP

3. 瓶颈分析

最初的架构在上线后，总体运行高效稳定，但是偶尔有一些TASK的处理时间会过久，我们针对这一现象进行了复现，通过复现我们发现当较多TASK集中在同一个消费者内部时，这些TASK的处理时间会更久。

有了这些观察，我们提出了猜测：Subscriber 同时处理的 TASK 越来越多，久而久之，线程池内的线程会越来越不够用，TASK 在同一 Subscriber 内的堆积，会导致了这批 TASK 的处理时延大幅上升。

为了验证这个猜测，在测试环境尝试了验证，测试环境和内容如下

配置: 8C16G
Subscriber WorkerThread 数：8

测试 TASK 的 STEP_CONFIG，共有3个STEP，step_2 中将 sleep 500ms，以模拟IO请求耗时，并且 step_2 会返回失败，然后重复执行10次，每次重试希望间隔3秒。考虑 step_1，step_2 只有简单的计算任务，这个 TASK 的期望执行时间应该是 0.5 + (3 + 0.5) * 10 = 35.5 秒左右。

step_1  # 记录开始时间
step_2  # sleep(0.5), 执行11次，间隔3秒
step_3  # 记录结束时间，计算 TASK 执行总时间

结合STEP_CONFIG和Subscriber的架构模型，我们可以得出Subscriber最佳性能的计算公式，当每个STEP的等待间隔时间内，线程能去做其他TASK的STEP，此时效率是最快的，如果STEP的期望间隔时间超过了实际间隔时间，此时TASK的处理效率是偏低的。由此公式可以算出，Subscriber最佳的TASK数量是56。

max_step_num ≈ (step_interval/step_cost + 1) * thread_num

我们测试并记录了单进程 Subscriber 同时处理不同数量 TASK 的平均处理时延，结果如下

实验数据

从图中可以看出，当 Subscriber 同时处理少于 50 个测试 TASK 时，可以保证每个 TASK 的平均处理时延接近 35.5 秒，当同时处理的 TASK 数量增加时，每个 TASK 的平均处理时延就急剧上升，当同时处理100个 TASK 时，TASK 的平均时长已经到了 78.827 秒，远超 35.5 秒。当超过TASK数量超过 56 时，TASK 的平均处理时长会大幅增加，这也是 Subscriber 的瓶颈所在。

4. 新架构

由于无法预知 TASK 的 workload，TASK 之间的work-sharing调度极有可能导致TASK倾斜，多个高workload的TASK集中某一个消费者进程内，超过消费者的性能阈值，进而导致TASK效率下降。那么有没有办法让 work-sharing 更平均一点呢？假设 Subscriber 最多同时处理 n 个 TASK，性能不会受到影响，我们可以参考TCP滑动窗口的思路，为每个 Subscriber 设置一个窗口值 window_size，假如当前 Subscriber 正在执行 m 个 TASK，window_size = n - m，当 window_size > 0 时，Subscriber 可以继续接收 TASK，window_size 等于 0 时，就暂停接收新的任务，专注完成已接收的任务，当已接收的任务完成时，window_size 扩大，则又可以接收新的 TASK。这样为消费者增加滑动窗口机制，做就能即保证公平的 work-sharing，而不造成消费者过载的现象。

增加滑动窗口帮助我们解决了 TASK 倾斜的问题，但是 Subscriber 假如异常退出，会导致执行中的 TASK 丢失。我们希望达到的效果是：任意的消费者进程挂掉，也许会导致系统服务降级，但绝不会导致执行中的 TASK 状态异常，而系统的服务降级可以通过消费者的重启或扩容恢复。Subscriber 进程对 TASK 的完整生命周期负责，导致了进程的“有状态化”，而“有状态化”正是引入高可用隐患的“始作俑者”，既然如此，只要将 Subscriber 改为“无状态化”的消费者，高可用的问题便解决了。

为了将“有状态化”改造成“无状态化”，我们对 Subscriber 进行了一些改造，开发了新的消费者 Accelerator，和 Subscriber 相比，Accelerator 有以下特点：

1. 从消费 TASK，转而消费 STEP
2. 配合 RabbitMQ，完成 STEP 在 Accelerator 之间的流转
3. 增加窗口机制，避免消费者过载

Accelerator 架构

Accelerator 仍然继承了 Subscriber 的 STEP-CONFIG 设计与 Reactor 模型，保证自身的灵活性与高吞吐，2者最大的区别在于 Accelerator 接收的不是 TASK，而是 STEP，引入 Accelerator 后 TASK 的处理逻辑如下：

1. 从 MQ 获取到 CreateCVMTask 的第二个 STEP，即VerifyInstance
2. Accelerator 从 MQ 获取到该消息，获取待执行的 STEP，然后将 STEP 执行相关的消息加入 Execute Queue
3. MasterThread 从队列中获取 STEP 执行相关的信息 exec_info
4. 如果要执行新的 STEP，为新的 STEP 创建事件(eventfd)，并加入事件循环；如果要重试 STEP，为当前 STEP 创建定时事件(timerfd)，加入事件循环
5. WorkerThread 收到事件后，获取该事件对应的 exec_info，然后开始工作
6. WorkerThread 完成工作后，将该 STEP 的结果加入 Execute Queue
7. MasterThread 从 Execute Queue 得到 STEP 结束的消息后，进行判断，如果 STEP 需要重试，则重新投入事件循环，在进程内重试，如果 STEP 成功或失败，则将接下来待执行的 STEP 也就是RecordResult加入 MQ，待其他 Accelerator 消费者获取到此消息，并完成RecordResult 这个 STEP。

可以看出，3-6步和 Subscriber 是非常接近的，区别在于 Accelerator 每完成一个 STEP，下一步的 STEP 会通过 RabbitMQ 进行投递，交给其他的 Accelerator 消费者完成。得益于RabbitMQ的ack机制，如果消费者未ack消息而意外退出，RabbitMQ会将未ack的消息重新投递给其他消费者，这样就保证了消费者进程的“无状态化”，任意Accelerator进程退出，都不会影响TASK的整体执行进展。

5. 效果

引入 Accelerator 之后，系统的 STEP 分布情况如下图

STEP 分布

理论上，将消息与状态相关联，让每一个步骤的状态都可以在框架层面上持久化了，未来如果有更高的可用性要求，可以借助其他的持久化工具来保存中间状态。

模拟之前 TASK 倾斜的场景，首先启动一个 Subscriber/Accelerator 进程1，下发 50 个 TASK，然后新启动一个 Subscriber/Accelerator 进程2，再次下发 50 个 TASK，并对比这 100 个 TASK 的处理情况。由于 TASK 分两次下发，会导致 Subscriber-1 进程接收 75 个左右的 TASK，而 Subscribe-2 分到 25 个左右的 TASK。实验结果如下图

对比数据

同样面对 TASK 倾斜的情况，Subscriber-1 和 Subscriber-2 的平均处理时长相差较大，Subscriber-2 的 TASK 平均在 36.775 秒左右完成，比 Subscriber-1 快了近10秒，而 Accelerator-1 和 Accelerator-2 则均在 37 秒左右完成，单单对比 Accelerator-1 和 Subscriber-1 的表现，Subscriber-1的耗时要多 26% 左右。

Accelerator 对比 Subscriber 增加的时间：MQ网络传输时间 * 步骤数量。增加了额外的网络开销，加大了 RabbitMQ 中消息流转的数量。

而这些额外的开销带来的好处：

• 均匀的 TASK 消费速度，避免 TASK 倾斜造成的性能下降
• 消费者无状态化，带来高可用性的提升，不用再怕混沌工程中的 ChaosMonkey
• 更加弹性，把每个步骤分在不同进程里，大大提升了可扩展性

总结

本文为大家介绍了一款基于腾讯云API二次开发的任务调度框架，此框架天然适应云的API，能快速、批量、稳定地完成各类云资源的操作任务。同时也对此框架的瓶颈进行了实验与分析，并提出了一种更优化的思路，然后对两种实现进行了定性和定量的比较。希望这套框架的思路能给准备使用腾讯云API进行二次开发的开发者们一点启发，欢迎大家多多交流。

参考

• 《让事件飞 ——Linux eventfd 原理与实践》
• 《可扩展的任务流框架实现（一）》
• https://www.man7.org/linux/man-pages/man2/eventfd.2.html
• https://man7.org/linux/man-pages/man2/timerfd_settime.2.html
• https://www.rabbitmq.com/consumer-prefetch.html
• https://subscription.packtpub.com/book/web_development/9781783287314/1/ch01lvl1sec09/the-reactor-pattern