2019年1月份接到一个需求，就是要把PHP相关的定时任务接到xxl里面。

我想大多数PHPer第一时间可能会问，为啥呀？

1、好管理，解放运维小哥

2、执行过程信息更明了

那么我们看看xxl-job这个java任务调度的架构设计图是怎样的？

关于xxl过多的我就不多说了，大家可以把源码下下来运行跑起来，代码实现清晰明白，好废话不多说，进入整体。链接：https://www.xuxueli.com/xxl-job/#%E4%BA%8C%E3%80%81%E5%BF%AB%E9%80%9F%E5%85%A5%E9%97%A8

大家是不是看完xxl-job 心里有个疑问，不是已经支持了php的glue模式吗？

不过我们有些定时任务不是单文件的脚本任务，而是基于框架实现的一些脚本任务，那这种的就没有办法通过glue模式来进行调度执行了。

所以就有了这款分布式高性能的xxl-job的php执行器当然不限于php脚本执行，它其实是一个rpc框架，是一款基于swoole网络框架为xxl-job量身打造的一款任务调度执行器，对应xxl-job的架构图里面的就是xxl-rpc

0x01 灵感

接到任务后，结合自己过往的开发经验，想到腾讯的这代码仓库：https://github.com/Tencent/tsf

有一些命令行交互体验还不错，比如：

[root@iz8vb25ut078wb3d93fcvmz xxl-job-swoole]# php Bin/index.php JobServer statusxxl-job-executor-server is not running, please check it [FAIL][root@iz8vb25ut078wb3d93fcvmz xxl-job-swoole]# php Bin/index.php JobServer stoproot     23531  0.0  0.5 296148  5464 ?        Ss   12:30   0:00 xxl-job-executor-serverroot     23531  0.0  0.5 296148  6036 ?        Ss   12:30   0:00 xxl-job-executor-serverstop xxl-job-executor-server [SUCCESS]

包含的命令：start, reload, stop, restart,status,list

然后就整体参考这里的代码进行实现。

0x02 功能梳理

然后就是对接xxl,一开始公司用的是二进制hessian协议进行通信，PHP基于这个协议，当时没有调通，最后采用的是json格式。xxl-job server那边采用了netty 作为通信框架，最终调通了后，现在开始梳理下需要实现的功能：

1、执行器需要实现主动触发的功能：

1. 任务结果回调服务；

2. 执行器注册服务；

3. 执行器注册摘除服务

4. 故障自动转移

5. 注册失败需要发起钉钉报警

6. 任务task worker与任务绑定

2、调度中心需要执行器提供的接口：

1. 心跳检测

2. 忙碌检测

3. 触发任务执行

4. 获取Rolling Log

5. 终止任务

6. 三种调度策略的实现【串行执行，丢弃后续调度，覆盖之前的调度】

0x03 通信消息协议

开始着手设计这套框架。

大致功能流程如下图：

rpc通信消息格式如下：

1、首先把收到的消息体转成字节数组

    /**     * 转换字符串到字节数组     *     * @param $str     * @return array     */   public static function convertStrToBytes($str){        $len = strlen($str);        $bytes = [];        for ($i = 0; $i < $len; $i++) {            if (ord($str[$i]) >= 128) {                $byte = ord($str[$i]) - 256;            } else {                $byte = ord($str[$i]);            }            $bytes[] = $byte;        }        return $bytes;    }

2、再把字符串数组转成截取前面四个，然后转成字符串，也就是json字符串

    array_slice($bytes, 4)

3、然后把json转成数组

    /**     * 转换字节数组到字符串     *     * @param $byteArr     * @return string     */    public static function convertByteArrToStr($byteArr){        $convertStr = '';        foreach ($byteArr as $byte) {            $convertStr .= chr($byte);        }        return $convertStr;    }

这样就把通信协议部分打通了，然后发送相关消息到xxl-job调度中心，就需要逆向组合数据了。

// 这需要逆向发给调度中心$message = ['result' => $result, 'requestId' => $req['requestId'], 'errorMsg' => null];

您可以想想，怎么弄？如需了解我的逆向写法，可以留言交流。

0x04 调度策略

串行：上一个任务没有执行完成，下一个就需要等待执行

丢弃后续调度：上一个任务没有执行完成，下一个任务丢弃调度。

覆盖之前的调度：上一个任务没有执行完成，下一个任务调度时，需要把之前的任务杀掉，然后该任务继续执行

丢弃后续调度，需要把下一个任务丢弃掉，然后给调度中心回调。

/** * @param $data * @return false|string */public static function discard($data, Table $cacheTable, $retryTimes = 5){// 被kill完回调while($retryTimes > 0) {            $hostInfo = self::getCurrentServer($cacheTable);            $bizCenter = self::getBizCenterByHostInfo($hostInfo);            $res = $bizCenter->callback(self::convertSecondToMicroS(), $data['logId'], $data['requestId'], $data['logDateTim'], ['code' => Code::SUCCESS_CODE,'msg' => '脚本被强杀，结束脚本运行',            ]);if ($res) {self::appendLog($data['logId'], 'callback回调:' . json_encode([$res]));self::appendLog($data['logId'], '任务因策略，没有进入执行');                $retryTimes = 0;            } else {self::appendLog($data['logId'], 'callback回调 retry:' . $retryTimes);                $retryTimes--;            }        }return true;    }

覆盖之前的，需要把上一个任务执行终止掉，然后发送回调给调度中心，下一个人继续执行

public static function coverEarly($existInfo, $data, Table $table, Table $cacheTable, $retryTimes = 5){        $process = new Process(function (Process $process) use ($data, $existInfo, $table, $cacheTable, $retryTimes) {// 杀掉之前            ExecutorCenter::kill($existInfo['jobId'], $table);// 被kill完回调while($retryTimes > 0) {                $hostInfo = self::getCurrentServer($cacheTable);self::appendLog($existInfo['log_id'], $hostInfo. 'callback回调:' . json_encode($existInfo));                $bizCenter = self::getBizCenterByHostInfo($hostInfo);                $res = $bizCenter->callback(self::convertSecondToMicroS(), $existInfo['log_id'], $existInfo['request_id'], $existInfo['log_date_time'], ['code' => Code::SUCCESS_CODE,'msg' => '脚本被强杀，结束脚本运行',                ]);if ($res) {self::appendLog($existInfo['log_id'], 'callback回调:' . json_encode([$res]));self::appendLog($existInfo['log_id'], '任务因策略，没有继续执行，已中断');                    $retryTimes = 0;                } else {                    $retryTimes--;                }            }        }, true);        $process->start();        $process->name('kill-'.$existInfo['log_id'] . '-process');    }

目前该项目还没有申请开源，也不知道有没有开源的价值，如果您也在用xxl-job作为任务调度中心，且目前仍然有PHP技术栈的话，可以留言交流。

如何做到串行执行且如何确保与其他任务不存在竞争task worker的呢？

答案是：task投递时指定目标task worker id

$server->task(['job_data' => $param, 'params' => $handleParams], $dstWorkerNum);

0x05 真实的项目调度

x-docker.ini

[server]; server type:tcp udp httptype = tcp; hosthost = 0.0.0.0; portport = 9506; process nameprocess_name = xxl-job-php-executor; registryapp_name = xxl-job-php; server nameserver_name = 'http://xxl-job-php.test.com'; phpphp = /usr/local/bin/php;shell = /bin/bash[setting]; worker process numworker_num = 2; Reactor numreactor_num = 8; task process numtask_worker_num = 20; dispatch modedispatch_mode = 2; daemonizedaemonize = 1; heartbeatheartbeat_check_interval = 60; idleheartbeat_idle_time = 600; system loglog_file = '/var/www/html/xxl-job-php-executor/Log/runtime.log'; mac process_name[table]size = 1024[project]root_path = '/var/www/html/'[xxljob]; site domainhost = docker.for.mac.host.internal; netty server portport = 8888; registry timemsregistry_interval_ms = 20000; a switch of registryopen_registry = 1; 备份netty server[xxljob_backup]; netty server，支持多个逗号拼接host_urls = docker.for.mac.host.internal:8889,docker.for.mac.host.internal:8890; registry timemsregistry_interval_ms = 20000; a switch of registryopen_registry = 1

支持类tp、类laravel框架,也可以扩展支持其他框架，很简单。配置如下：

'commonmode' => [     'router_index' => '/cli.php',     'file_real_path' => ':string/Application/:string/Controller/:stringController.class.php',     'identifier' => 't'],'artisanmode' => [     'run_mode' => self::ARTISAN_MODE,     'router_index' => self::ARTISAN_MODE,     'param_identify' => ' ',     'file_real_path' => ':string/Console/Kernel.php',     'identifier' => 'c']

0x06 总结

通过此项目的开发，可以完全利用swoole来进行网络编程开发，swoole的优质特性，值得广为使用。在过程中，对一些swoole的功能特性有较深入的理解，从头开始实现一个rpc框架很容易，然后就把精力集中放在实现相关功能上，task 的用法在低版本有些不是很好，所以如果是用的php5.6版本的，可以考虑用1.10.5的版本，对此项目来说，足够了

0x07 引用

https://www.php.net/manual/en/pcntl.constants.php

https://wiki.swoole.com/wiki/page/1.html

https://wiki.swoole.com/#/

https://github.com/Tencent/tsf

感谢swoole 项目开发组提供的如此优秀的扩展项目，感谢tencent tsf项目提供的灵感。

如果这篇文章让你有一点点收获，点下在看哦