@yjc/single-task

概述

这是一个触发式的任务调度模块，通过配置文件声明任务的执行命令、默认参数和任务之间的依赖。

插入一个新任务A时，检查依赖。 如果A依赖的任务正在执行，则A进入等待队列；如果有其它任务依赖A，则自动把依赖A的任务插入等待队列。 任何一个任务执行完毕，或者新任务插入后，遍历等待队列、排重，调起没有依赖的任务。

此模块基于SOCKET服务，任务队列和日志依赖MYSQL存储。支持单服务器任务调度。

使用准备

在MYSQL服务端新建2个表，分别用于日志记录和任务队列存储。表结构见readme/table.sql。

启动

node @yjc/single-task/index.js serverConfigModulePath

配置文件参考示例test/server.config.js和test/task.config.js

服务启动后，监听指定的端口，响应SOCKET连接进行通信。

示例

建表后，修改数据库配置文件test/db.json。然后执行

cd @yjc/single-task
node index.js test/server.config.js

可启动测试服务端。

客户端示例见test/client.js和test/client.php。

错误告警和重试机制

如果任务进程异常终止，或者退出代码 (exit code) 不为0，则认为任务出错。 任务出错时会阻塞所有下游任务，同时调起告警任务（如果配置了taskConfig.alert）。

如果出现了预期错误，需要重试任务，发送代码1001到父进程：

process.send && process.send(1001);

如果发送不等于1001的非负整数，也认为任务出错。

任务触发

调度服务调起任务进程时，开启了IPC通道。子进程中可向父进程发送数据。例如：

if (process.send) {
    process.send([
        cmd: 'taskName'
    ]);
}

可调起新的任务。

任务配置

{String} logDir

任务进程输出流所在目录。

{Integer} maxRetryTimes

最大重试次数。此属性可配置到具体的任务。

{String} alter

告警任务。

{Object} task

详细任务配置。

详细任务配置

dep.${depCmd}.mapArgs(args)

next.${nextCmd}.mapArgs(args)

• args {Array} 上游任务执行时使用的参数列表
• return {Array|null}

返回一个数组，数组的每个元素都是一个参数列表数组。 上游任务执行完成后，调起N(=returnArray.length)个下游任务。 返回值的每个元素作为每个下游任务的参数列表。

如果返回null，则不自动调起下游任务。

duplicateArgs(argsList)

• argsList {Array} 将要执行的任务的参数列表数组
• return {Array}

任务调度允许同一个命令的不同传参同时执行。此配置函数筛选将要执行的所有参数列表，把参数列表不同、但实际上冲突或重复的参数列表排除。

返回值的每个元素表示要排除的argsList的元素的索引值。

协议

1. 服务端接收的字符串，读取头部4个字节，按大字节序（BE）解析为整数，视为随后等待接收的总字节数

2. 服务端接收的字符串，去掉头部4个字节，解析后得到一个非空对象，该对象包含以下属性：

   • auth {String} 身份验证字符串
   • action {String} 指令
     • touch 检查服务是否运行
     • new tasks 插入新任务
     • pid running status 进程ID对应的任务是否正在执行（1:正在执行;0:未执行）
     • queue status 取得任务队列信息
     • test task config 测试任务配置文件
     • reload 等待正在执行的任务执行完成后，重启服务
     • restart 强制重启服务
     • exec 强制执行任务队列，正常情况下不需要使用此指令
   • namespace {String} 命令所属的命名空间
   • cmd {String} 待执行的命令
   • args {Array|null} 命令参数
   • config {String} 任务配置文件的路径
   • data {Array} 列表数据

3. 服务端响应，error属性表示出错信息，result属性表示结果，序列化后，在头部拼接序列化字节长度（4字节的大字节序整数）

默认处理

1. 解析方法：JSON.parse
2. 身份验证方法：time + '|' + md5(time + authKey)
3. 序列化方法：JSON.stringify