Swoft源碼之Swoole和Swoft的分析

//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);

    }

}

//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;

    }

}

//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;

    }

}

//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);

}

//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);

}

//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\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;

}