Swooft의 Swoole 및 Swoft 소스 코드 분석-PHP 튜토리얼-php.cn

이 기사의 내용은 Swoft 소스 코드 분석(작업 전달/예약 작업)에 대해 Swoole 및 Swoft에 대한 소개를 공유하는 것입니다. 필요한 친구가 참조할 수 있습니다.

Preface

Swoft의 작업 기능은 Swoole의 작업 메커니즘을 기반으로 합니다. >, 즉 Swoft의 Task 메커니즘은 본질적으로 Swoole의 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) <p><code>swoft.pipeMessage</code>事件最终由<code>PipeMessageListener</code>处理。在相关的监听其中，如果发现<code>swoft.pipeMessage</code>事件由<code>Task::deliverByProcess()</code>产生的，<code>Worker进程</code>会替其执行一次<code>Task::deliver()</code>，最终将任务数据投递到<code>TaskWorker进程</code>中。</p><p>一道简单的回顾练习：从<code>Task::deliverByProcess()</code>到某<code>TaskBean</code></p>Task 전달#🎜🎜#<div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:php;toolbar:false">//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;
}

로그인 후 복사

위 내용은 Swooft의 Swoole 및 Swoft 소스 코드 분석의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!