当前位置：移动技术网 > IT编程>开发语言>PHP > Swoft源码之Swoole和Swoft的分析

Swoft源码之Swoole和Swoft的分析

2019年11月07日 | 移动技术网IT编程 | 我要评论

这篇文章给大家分享的内容是关于swoft 源码剖析之swoole和swoft的一些介绍(task投递/定时任务篇），有一定的参考价值，有需要的朋友可以参考一下。

前言

swoft的任务功能基于swoole的task机制，或者说swoft的task机制本质就是对swoole的task机制的封装和加强。

任务投递

//swoft\task\task.php
class task
{
    /**
     * deliver coroutine or async task
     *
     * @param string $taskname
     * @param string $methodname
     * @param array  $params
     * @param string $type
     * @param int    $timeout
     *
     * @return bool|array
     * @throws taskexception
     */
    public static function deliver(string $taskname, string $methodname, array $params = [], string $type = self::type_co, $timeout = 3)
    {
        $data   = taskhelper::pack($taskname, $methodname, $params, $type);
 
        if(!app::isworkerstatus() && !app::iscocontext()){
            return self::deliverbyqueue($data);//见下文command章节
        }
 
        if(!app::isworkerstatus() && app::iscocontext()){
            throw new taskexception('please deliver task by http!');
        }
 
 
        $server = app::$server->getserver();
        // delier coroutine task
        if ($type == self::type_co) {
            $tasks[0]  = $data;
            $priflekey = 'task' . '.' . $taskname . '.' . $methodname;
 
            app::profilestart($priflekey);
            $result = $server->taskco($tasks, $timeout);
            app::profileend($priflekey);
 
            return $result;
        }
 
        // deliver async task
        return $server->task($data);
    }
}

任务投递task::deliver()将调用参数打包后根据$type参数通过swoole的$server->taskco()或$server->task()接口投递到task进程。
task本身始终是同步执行的，$type仅仅影响投递这一操作的行为，task::type_async对应的$server->task()是异步投递，task::deliver()调用后马上返回；task::type_co对应的$server->taskco()是协程投递，投递后让出协程控制，任务完成或执行超时后task::deliver()才从协程返回。

任务执行

//swoft\task\bootstrap\listeners\taskeventlistener 
/**
 * the listener of swoole task
 * @swoolelistener({
 *     swooleevent::on_task,
 *     swooleevent::on_finish,
 * })
 */
class taskeventlistener implements taskinterface, finishinterface
{
    /**
     * @param \swoole\server $server
     * @param int            $taskid
     * @param int            $workerid
     * @param mixed          $data
     * @return mixed
     * @throws \invalidargumentexception
     */
    public function ontask(server $server, int $taskid, int $workerid, $data)
    {
        try {
            /* @var taskexecutor $taskexecutor*/
            $taskexecutor = app::getbean(taskexecutor::class);
            $result = $taskexecutor->run($data);
        } catch (\throwable $throwable) {
            app::error(sprintf('taskexecutor->run %s file=%s line=%d ', $throwable->getmessage(), $throwable->getfile(), $throwable->getline()));
            $result = false;
 
            // release system resources
            app::trigger(appevent::resource_release);
 
            app::trigger(taskevent::after_task);
        }
        return $result;
    }
}

此处是swoole.ontask的事件回调，其职责仅仅是将将worker进程投递来的打包后的数据转发给taskexecutor。

swoole的task机制的本质是worker进程将耗时任务投递给同步的task进程(又名taskworker)处理,所以swoole.ontask的事件回调是在task进程中执行的。上文说过,worker进程是你大部分http服务代码执行的环境，但是从taskeventlistener.ontask()方法开始，代码的执行环境都是task进程，也就是说，taskexecutor和具体的taskbean都是执行在task进程中的。

//swoft\task\taskexecutor
/**
 * the task executor
 *
 * @bean()
 */
class taskexecutor
{
    /**
     * @param string $data
     * @return mixed
    */
    public function run(string $data)
    {
        $data = taskhelper::unpack($data);
 
        $name   = $data['name'];
        $type   = $data['type'];
        $method = $data['method'];
        $params = $data['params'];
        $logid  = $data['logid'] ?? uniqid('', true);
        $spanid = $data['spanid'] ?? 0;
 
 
        $collector = taskcollector::getcollector();
        if (!isset($collector['task'][$name])) {
            return false;
        }
 
        list(, $coroutine) = $collector['task'][$name];
        $task = app::getbean($name);
        if ($coroutine) {
            $result = $this->runcotask($task, $method, $params, $logid, $spanid, $name, $type);
        } else {
            $result = $this->runsynctask($task, $method, $params, $logid, $spanid, $name, $type);
        }
 
        return $result;
    }
}

任务执行思路很简单，将worker进程发过来的数据解包还原成原来的调用参数，根据$name参数找到对应的taskbean并调用其对应的task()方法。其中taskbean使用类级别注解@task(name="taskname")或者@task("taskname")声明。

值得一提的一点是，@task注解除了name属性，还有一个coroutine属性，上述代码会根据该参数选择使用协程的runcotask()或者同步的runsynctask()执行task。但是由于而且由于swoole的task进程的执行是完全同步的，不支持协程，所以目前版本请该参数不要配置为true。同样的在taskbean中编写的任务代码必须的同步阻塞的或者是要能根据环境自动将异步非阻塞和协程降级为同步阻塞的

从process中投递任务

前面我们提到：

swoole的task机制的本质是worker进程将耗时任务投递给同步的task进程(又名 taskworker)处理。

换句话说，swoole的$server->taskco()或$server->task()都只能在worker进程中使用。
这个限制大大的限制了使用场景。如何能够为了能够在process中投递任务呢？swoft为了绕过这个限制提供了task::deliverbyprocess()方法。其实现原理也很简单，通过swoole的$server->sendmessage()方法将调用信息从process中投递到worker进程中，然后由worker进程替其投递到task进程当中，相关代码如下：

//swoft\task\task.php
/**
 * deliver task by process
 *
 * @param string $taskname
 * @param string $methodname
 * @param array  $params
 * @param string $type
 * @param int    $timeout
 * @param int    $workid
 *
 * @return bool
 */
public static function deliverbyprocess(string $taskname, string $methodname, array $params = [], int $timeout = 3, int $workid = 0, string $type = self::type_async): bool
{
    /* @var pipemessageinterface $pipemessage */
    $server      = app::$server->getserver();
    $pipemessage = app::getbean(pipemessage::class);
    $data = [
        'name'    => $taskname,
        'method'  => $methodname,
        'params'  => $params,
        'timeout' => $timeout,
        'type'    => $type,
    ];
 
    $message = $pipemessage->pack(pipemessage::message_type_task, $data);
    return $server->sendmessage($message, $workid);
}

数据打包后使用$server->sendmessage()投递给worker:

//swoft\bootstrap\server\servertrait.php
/**
 * onpipemessage event callback
 *
 * @param \swoole\server $server
 * @param int            $srcworkerid
 * @param string         $message
 * @return void
 * @throws \invalidargumentexception
 */
public function onpipemessage(server $server, int $srcworkerid, string $message)
{
    /* @var pipemessageinterface $pipemessage */
    $pipemessage = app::getbean(pipemessage::class);
    list($type, $data) = $pipemessage->unpack($message);
 
    app::trigger(appevent::pipe_message, null, $type, $data, $srcworkerid);
}

$server->sendmessage后,worker进程收到数据时会触发一个swoole.pipemessage事件的回调，swoft会将其转换成自己的swoft.pipemessage事件并触发.

//swoft\task\event\listeners\pipemessagelistener.php
/**
 * the pipe message listener
 *
 * @listener(event=appevent::pipe_message)
 */
class pipemessagelistener implements eventhandlerinterface
{
    /**
     * @param \swoft\event\eventinterface $event
     */
    public function handle(eventinterface $event)
    {
        $params = $event->getparams();
        if (count($params) < 3) {
            return;
        }
 
        list($type, $data, $srcworkerid) = $params;
 
        if ($type != pipemessage::message_type_task) {
            return;
        }
 
        $type       = $data['type'];
        $taskname   = $data['name'];
        $params     = $data['params'];
        $timeout    = $data['timeout'];
        $methodname = $data['method'];
 
        // delever task
        task::deliver($taskname, $methodname, $params, $type, $timeout);
    }
}

swoft.pipemessage事件最终由pipemessagelistener处理。在相关的监听其中，如果发现swoft.pipemessage事件由task::deliverbyprocess()产生的，worker进程会替其执行一次task::deliver()，最终将任务数据投递到taskworker进程中。

一道简单的回顾练习：从task::deliverbyprocess()到某taskbean 最终执行任务，经历了哪些进程，而调用链的哪些部分又分别是在哪些进程中执行？

从command进程或其子进程中投递任务

//swoft\task\queuetask.php
/**
 * @param string $data
 * @param int    $taskworkerid
 * @param int    $srcworkerid
 *
 * @return bool
 */
public function deliver(string $data, int $taskworkerid = null, $srcworkerid = null)
{
    if ($taskworkerid === null) {
        $taskworkerid = mt_rand($this->workernum + 1, $this->workernum + $this->tasknum);
    }
 
    if ($srcworkerid === null) {
        $srcworkerid = mt_rand(0, $this->workernum - 1);
    }
 
    $this->check();
    $data   = $this->pack($data, $srcworkerid);
    $result = \msg_send($this->queueid, $taskworkerid, $data, false);
    if (!$result) {
        return false;
    }
 
    return true;
}

对于command进程的任务投递，情况会更复杂一点。
上文提到的process，其往往衍生于http/rpc服务，作为同一个manager的子孙进程，他们能够拿到swoole\server的句柄变量，从而通过$server->sendmessage(),$server->task()等方法进行任务投递。

但在swoft的体系中，还有一个十分路人的角色： command。
command的进程从shell或cronb独立启动，和http/rpc服务相关的进程没有亲缘关系。因此command进程以及从command中启动的process进程是没有办法拿到swoole\server的调用句柄直接通过unixsocket进行任务投递的。
为了为这种进程提供任务投递支持，swoft利用了swoole的task进程的一个特殊功能----消息队列。

使用消息队列的task进程.png

同一个项目中command和http\rpcserver 通过约定一个message_queue_key获取到系统内核中的同一条消息队列，然后comand进程就可以通过该消息队列向task进程投递任务了。
该机制没有提供对外的公开方法，仅仅被包含在task::deliver()方法中，swoft会根据当前环境隐式切换投递方式。但该消息队列的实现依赖semaphore拓展，如果你想使用，需要在编译php时加上--enable-sysvmsg参数。

定时任务

除了手动执行的普通任务，swoft还提供了精度为秒的定时任务功能用来在项目中替代linux的crontab功能.

swoft用两个前置process---任务计划进程：crontimerprocess和任务执行进程cronexecprocess
，和两张内存数据表-----runtimetable（任务(配置)表）origintable（（任务）执行表）用于定时任务的管理调度。
两张表的每行记录的结构如下:

\\swoft\task\crontab\tablecrontab.php
/**
 * 任务表，记录用户配置的任务信息
 * 表每行记录包含的字段如下,其中`rule`,`taskclass`,`taskmethod`生成key唯一确定一条记录
 * @var array $originstruct 
 */
private $originstruct = [
    'rule'       => [\swoole\table::type_string, 100],//定时任务执行规则，对应@scheduled注解的cron属性
    'taskclass'  => [\swoole\table::type_string, 255],//任务名 对应@task的name属性(默认为类名)
    'taskmethod' => [\swoole\table::type_string, 255],//task方法，对应@scheduled注解所在方法
    'add_time'   => [\swoole\table::type_string, 11],//初始化该表内容时的10位时间戳
];
 
/**
 * 执行表，记录短时间内要执行的任务列表及其执行状态
 * 表每行记录包含的字段如下,其中`taskclass`,`taskmethod`,`minute`,`sec`生成key唯一确定一条记录
 * @var array $runtimestruct 
 */
private $runtimestruct = [
    'taskclass'  => [\swoole\table::type_string, 255],//同上
    'taskmethod' => [\swoole\table::type_string, 255],//同上
    'minute'      => [\swoole\table::type_string, 20],//需要执行任务的时间，精确到分钟 格式date('ymdhi')
    'sec'        => [\swoole\table::type_string, 20],//需要执行任务的时间，精确到分钟 10位时间戳
    'runstatus'  => [\swoole\table::type_int, 4],//任务状态，有 0(未执行)  1(已执行)  2（执行中） 三种。 
    //注意：这里的执行是一个容易误解的地方，此处的执行并不是指任务本身的执行，而是值`任务投递`这一操作的执行，从宏观上看换成 _未投递_,_已投递_，_投递中_描述会更准确。
];

此处为何要使用swoole的内存table?

swoft的的定时任务管理是分别由 任务计划进程 和 任务执行进程 进程负责的。两个进程的运行共同管理定时任务，如果使用进程间独立的array()等结构，两个进程必然需要频繁的进程间通信。而使用跨进程的table(本文的table,除非特别说明，都指swoole的swoole\table结构)直接进行进程间数据共享，不仅性能高，操作简单还解耦了两个进程。

为了table能够在两个进程间共同使用,table必须在swoole server启动前创建并分配内存。具体代码在swoft\task\bootstrap\listeners->onbeforestart()中，比较简单，有兴趣的可以自行阅读。

背景介绍完了，我们来看看这两个定时任务进程的行为

//swoft\task\bootstrap\process\crontimerprocess.php
/**
 * crontab timer process
 *
 * @process(name="crontimer", boot=true)
 */
class crontimerprocess implements processinterface
{
    /**
     * @param \swoft\process\process $process
     */
    public function run(swoftprocess $process)
    {
        //code....
        /* @var \swoft\task\crontab\crontab $cron*/
        $cron = app::getbean('crontab');
 
        // swoole/httpserver
        $server = app::$server->getserver();
 
        $time = (60 - date('s')) * 1000;
        $server->after($time, function () use ($server, $cron) {
            // every minute check all tasks, and prepare the tasks that next execution point needs
            $cron->checktask();
            $server->tick(60 * 1000, function () use ($cron) {
                $cron->checktask();
            });
        });
    }
}

//swoft\task\crontab\crontab.php
/**
 * 初始化runtimetable数据
 *
 * @param array $task        任务
 * @param array $parseresult 解析crontab命令规则结果，即task需要在当前分钟内的哪些秒执行
 * @return bool
 */
private function initruntimetabledata(array $task, array $parseresult): bool
{
    $runtimetabletasks = $this->getruntimetable()->table;
 
    $min = date('ymdhi');
    $sec = strtotime(date('y-m-d h:i'));
    foreach ($parseresult as $time) {
        $this->checktaskqueue(false);
        $key = $this->getkey($task['rule'], $task['taskclass'], $task['taskmethod'], $min, $time + $sec);
        $runtimetabletasks->set($key, [
            'taskclass'  => $task['taskclass'],
            'taskmethod' => $task['taskmethod'],
            'minute'     => $min,
            'sec'        => $time + $sec,
            'runstatus'  => self::normal
        ]);
    }
 
    return true;
}

crontimerprocess是swoft的定时任务调度进程，其核心方法是crontab->initruntimetabledata()。
该进程使用了swoole的定时器功能，通过swoole\timer在每分钟首秒时执行的回调，crontimerprocess每次被唤醒后都会遍历任务表计算出当前这一分钟内的60秒分别需要执行的任务清单，写入执行表并标记为未执行。

//swoft\task\bootstrap\process
/**
 * crontab process
 *
 * @process(name="cronexec", boot=true)
 */
class cronexecprocess implements processinterface
{
    /**
     * @param \swoft\process\process $process
     */
    public function run(swoftprocess $process)
    {
        $pname = app::$server->getpname();
        $process->name(sprintf('%s cronexec process', $pname));
 
        /** @var \swoft\task\crontab\crontab $cron */
        $cron = app::getbean('crontab');
 
        // swoole/httpserver
        $server = app::$server->getserver();
 
        $server->tick(0.5 * 1000, function () use ($cron) {
            $tasks = $cron->getexectasks();
            if (!empty($tasks)) {
                foreach ($tasks as $task) {
                    // diliver task
                    task::deliverbyprocess($task['taskclass'], $task['taskmethod']);
                    $cron->finishtask($task['key']);
                }
            }
        });
    }
}

cronexecprocess作为定时任务的执行者，通过swoole\timer每0.5s唤醒自身一次，然后把 执行表 遍历一次，挑选当下需要执行的任务，通过sendmessage()投递出去并更新该任务执行表中的状态。
该执行进程只负责任务的投递，任务的实际实际执行仍然在task进程中由taskexecutor处理。

定时任务的宏观执行情况如下：

您可能感兴趣的文章:

如对本文有疑问，点击进行留言回复！！

MongoDB $type 操作符

描述在本章节中，我们将继续讨论MongoDB中条件操作符 $type。$type操作符是基于BSON类型来检索集... [阅读全文]
彻底解决小程序无法触发SESSION问题

转自冯奎博客fengkui.net一、首先找到第一次发起网络请求的地址，将服务器返回set-cookie当全局变... [阅读全文]
tp5框架基于ajax实现异步删除图片的方法示例

本文实例讲述了tp5框架基于ajax实现异步删除图片的方法。分享给大家供大家参考，具体如下：为了提高用户体验，我们为商品相册制作了ajax无刷新异步删除的功能，... [阅读全文]
thinkphp5框架路由原理与用法详解

本文实例讲述了thinkphp5框架路由原理与用法。分享给大家供大家参考，具体如下：路由理解概括的说：路由就是网络请求的url与thinkphp应用层的逻辑处理... [阅读全文]
让whoops帮我们告别ThinkPHP6的异常页面

春节期间熟悉了tp6，也写了一个tp6的博客程序，但系统的异常页面实在另外头疼，很多时候无法查看到是哪行代码出的问题。所以就特别的想把whoops引进来，经过... [阅读全文]
thinkphp5 + ajax 使用formdata提交数据（包括文件上传）后台返回json完整实例

本文实例讲述了thinkphp5 + ajax 使用formdata提交数据（包括文件上传）后台返回json。分享给大家供大家参考，具体如下：知识点总结1.j... [阅读全文]
YII2框架实现表单中上传单个文件的方法示例

本文实例讲述了yii2框架实现表单中上传单个文件的方法。分享给大家供大家参考，具体如下：有些时候我们提交的表单中含有文件。怎么样让表单里的数据和文件一起提交。我... [阅读全文]
YII2框架中自定义用户认证模型，完成登陆和注册操作示例

本文实例讲述了yii2框架中自定义用户认证模型，完成登陆和注册操作。分享给大家供大家参考，具体如下：有些时候我们需要自已定义用户类，操作自已建的用户表，来完成登... [阅读全文]
YII2框架中分页组件的使用方法示例

本文实例讲述了yii2框架中分页组件的使用方法。分享给大家供大家参考，具体如下：当数据过多，无法一页显示时，我们经常会用到分页组件，yii2中已经帮我们封装好了... [阅读全文]
YII2框架中验证码的简单使用方法示例

本文实例讲述了yii2框架中验证码的简单使用方法。分享给大家供大家参考，具体如下：验证码的使用是比较频繁的。yii2中已经帮我们做好了封装。首先我们在控制器里创... [阅读全文]