处理异步任务(在 task worker 进程内被调用)
@param  Server $server
@param  int    $taskId
@param  int    $fromId
@param  mixed  $data
@return mixed
517247
public function onTask(Server $server, int $taskId, $fromId, $data)
    {
        $this->log('task worker received a new task', [
            'taskId'   => $taskId,
            'fromId'   => $fromId,
            'workerId' => $server->worker_id,
            'data'     => $data
        ], 'debug');

        // 返回任务执行的结果(finish操作是可选的，也可以不返回任何结果)
        return 'OK';
    }
650092
public function getTask($taskId)
    {
        $filename = $this->taskIdToFileName($taskId);
        if (!file_exists($filename)) {
            return null;
        }

        return $this->createTaskFromMetaData(new JsonFile($filename, null));
    }
1070002
public function run($taskId, $operation, array $arguments)
    {

        try {

            $lambda = new Lambda($operation, $this->ioc);

            if ($lambda->isInstanceMethod()) {
                $this->connectHooks($taskId, $lambda->getCallInstance());
            }

            $this->taskRepository->write($taskId, ['state' => Queue::RUNNING]);

            $result = $this->performTask($lambda, $arguments);

            $this->taskRepository->write($taskId, ['state' => Queue::FINISHED]);

            return $result;

        } catch (\Exception $e) {

            $this->taskRepository->writeMessage($taskId, $this->exceptionMessage($e), Chatty::FATAL);
            $this->taskRepository->write($taskId, ['state' => Queue::FAILED]);

        }

    }
977317
public function kickUid($uid, $fromDispatch = false)
    {
        if ($this->uid_fd_table->exist($uid)) {//本机处理
            $fd = $this->uid_fd_table->get($uid)['fd'];
            $this->close($fd);
        } else {
            if ($fromDispatch) {
                return;
            }
            if ($this->isCluster()) {
                ProcessManager::getInstance()->getRpcCall(ClusterProcess::class, true)->my_kickUid($uid);
            }
        }
    }
255468
public static function onWorkerStart($serv, $worker_id) {
		Yesf::app()->initInWorker();
		//根据类型，设置不同的进程名
		if ($serv->taskworker) {
			self::setProcessName(Yesf::app()->getConfig('name', Yesf::CONF_PROJECT) . ' task ' . $worker_id);
		} else {
			self::initHotReload($serv);
			self::setProcessName(Yesf::app()->getConfig('name', Yesf::CONF_PROJECT) . ' worker ' . $worker_id);
		}
		//清除opcache
		if (function_exists('opcache_reset')) {
			opcache_reset();
		}
		//标记一下
		Swoole::$isTaskWorker = $serv->taskworker;
		//回调
		Internal::onWorkerStart();
		Plugin::trigger('workerStart', [$serv->taskworker, $worker_id]);
	}
783228
public function runTask($taskId, $context = array())
    {
        $param =  array(
            'actionName' => 'runTask',
            'authPass'   => $this->getConfig('password'),
            'authUser'   => $this->getConfig('login'),
            'taskId'     => $taskId,
            'mode'       => 'synchronous'
        );

        if ($this->getConfig('context') != '' && is_array($this->getConfig('context'))) {
            $context = array_merge($this->getConfig('context'), $context);
        }

        if (!empty($context) && is_array($context)) {
            $param['context'] = $context;
        }

        return $this->getContent($this->doRequest($this->get('?'. $this->getJsonEncoded($param))));
    }
1223190
public function onReceive(SwServer $server, $fd, $fromId, $data): void
    {
        $data = trim($data);
        $this->log("Receive data [$data] from client [FD:$fd].");
        $server->send($fd, "I have been received your message.\n");

        // $this->onTaskReceive($server, $fd, $fromId, $data);

        // 群发收到的消息
        // $this->reloadWorker->write($data);
    }
1032647
public function getTask(int $taskId)
    {
        if ( ! $this->validateId($taskId)) {
            return false;
        }

        $result = $this->get('tasks/' . $taskId);

        $status = $result->getStatusCode();
        if (200 === $status) {
            return json_decode($result->getBody()->getContents());
        }

        return false;
    }
272961
public static function nPost(string $name, $data, $workerID = -1)
    {
        $task = TaskParser::getInstance()->getTask($name);
        $paramClass = $task['Task']->paramClass;
        return static::post(new TaskInfo(new $task['className'], new $paramClass($data)), $workerID);
    }
1163185
private void readWBS(ChildTaskContainer parent, Integer id)
   {
      Integer currentID = id;
      Table table = getTable("WBSTAB");

      while (currentID.intValue() != 0)
      {
         MapRow row = table.find(currentID);
         Integer taskID = row.getInteger("TASK_ID");
         Task task = readTask(parent, taskID);
         Integer childID = row.getInteger("CHILD_ID");
         if (childID.intValue() != 0)
         {
            readWBS(task, childID);
         }
         currentID = row.getInteger("NEXT_ID");
      }
   }
607890
public function onTask($serv, $taskId, $fromId, $data)
    {
        if (is_string($data)) {
            $unserializeData = unserialize($data);
        } else {
            $unserializeData = $data;
        }
        $type    = $unserializeData['type'] ?? '';
        $message = $unserializeData['message'] ?? '';
        $result  = false;
        $status  = 200;
        switch ($type) {
            case Macro::SERVER_TYPE_TASK:
                try {
                    $taskName      = $message['task_name'];
                    $taskFucName   = $message['task_fuc_name'];
                    $taskData      = $message['task_fuc_data'];
                    /**
                     * @var Context $taskContext
                     */
                    $taskContext   = $message['task_context'];
                    $taskConstruct = $message['task_construct'];

                    // 构造请求日志对象
                    $PGLog                            = clone getInstance()->log;
                    $PGLog->logId                     = $taskContext->getLogId();
                    $PGLog->accessRecord['beginTime'] = microtime(true);
                    $PGLog->accessRecord['uri']       = $taskContext->getInput()->getPathInfo();
                    $PGLog->pushLog('task', $taskName);
                    $PGLog->pushLog('method', $taskFucName);
                    defined('SYSTEM_NAME') && $PGLog->channel = SYSTEM_NAME . '-task';
                    $PGLog->init();
                    $taskContext->setLogId($PGLog->logId);
                    $taskContext->setLog($PGLog);

                    if (empty($taskName) || empty($taskFucName)) {
                        $status = 500;
                        $PGLog->error('Task Not Found');
                        $PGLog->pushLog('status', $status);
                        $PGLog->appendNoticeLog();
                        return $result;
                    }

                    $objectPool    = AOPFactory::getObjectPool($this->objectPool, $this);
                    /**
                     * @var Task $task
                     */
                    $task          = $objectPool->get($taskName, $taskConstruct);

                    // 运行方法
                    if (method_exists($task, $taskFucName)) {
                        //给task做初始化操作
                        $task->__initialization($taskId, $this->server->worker_pid, $taskName, $taskFucName, $taskContext, $objectPool);
                        $result = $task->$taskFucName(...$taskData);
                    } else {
                        throw new Exception("method $taskFucName not exist in $taskName");
                    }
                } catch (\Throwable $e) {
                    $status = 500;
                    $PGLog->error(dump($e, false, true));
                } finally {
                    $PGLog->pushLog('status', $status);

                    if ($status === 200) {
                        (mt_rand(0, 99) < $this->taskLogRate) && $PGLog->appendNoticeLog();
                    } else {
                        $PGLog->appendNoticeLog();
                    }

                    //销毁对象
                    foreach ($this->objectPoolBuckets as $k => $obj) {
                        $objectPool->push($obj);
                        $this->objectPoolBuckets[$k] = null;
                        unset($this->objectPoolBuckets[$k]);
                    }
                }
                return $result;
            default:
                return parent::onTask($serv, $taskId, $fromId, $data);
        }
    }
70922
function onReceive($serv, $client_id, $from_id, $data)
    {
        //检测request data完整性
        $ret = $this->checkData($client_id, $data);
        switch($ret)
        {
            //错误的请求
            case self::ST_ERROR;
                $this->server->close($client_id);
                return;
            //请求不完整，继续等待
            case self::ST_WAIT:
                return;
            default:
                break;
        }
        //完整的请求
        //开始处理

        /**
         * @var $request Swoole\Request
         */
        $request = $this->requests[$client_id];

        $request->fd = $client_id;
        $request->time = time();

        /**
         * Socket连接信息
         */
	    $info = $serv->connection_info($client_id);
        $request->server['SWOOLE_CONNECTION_INFO'] = $info;
        $request->remote_ip = $info['remote_ip'];
        $request->remote_port = $info['remote_port'];
        /**
         * Server变量
         */
        $request->server['REQUEST_URI'] = $request->meta['uri'];
        $request->server['REMOTE_ADDR'] = $request->remote_ip;
        $request->server['REMOTE_PORT'] = $request->remote_port;
        $request->server['REQUEST_METHOD'] = $request->meta['method'];
        $request->server['REQUEST_TIME'] = $request->time;
        $request->server['SERVER_PROTOCOL'] = $request->meta['protocol'];
        if (!empty($request->meta['query']))
        {
            $_SERVER['QUERY_STRING'] = $request->meta['query'];
        }
        $request->setGlobal();
        $this->parseRequest($request);
        $this->currentRequest = $request;
        //处理请求，产生response对象
        $response = $this->onRequest($request);
        if ($response and $response instanceof Swoole\Response)
        {
            //发送response
            $this->response($request, $response);
        }
    }
607422
private function runTask($workflow, $taskServiceId, $taskId)
    {
        $task = $this->container->get("{$workflow}.task.{$taskServiceId}");

        if (!$task instanceof ExecutableTaskInterface) {
            throw new InvalidTaskClassException(
                "The service {$taskServiceId} must implement the ExecutableTaskInterface interface."
            );
        }

        if ($task instanceof ContainerAwareInterface) {
            $task->setContainer($this->container);
        }

        if ($task instanceof LoggerAwareInterface) {
            /** @var LoggerInterface $logger */
            $logger = $this->container->get('logger');
            $task->setLogger($logger);
        }

        $parameters = new ParameterBag();
        $parameterBagId = "{$workflow}{$taskId}";

        /** @var ProviderInterface $stateStorage */
        $stateStorage = $this->container->get('state_storage');

        // Try to load the state file.
        try {
            $stateStorage->load($parameterBagId, $parameters);
        } catch (IOException $ignored) {
        }

        if ($task instanceof ParameterAwareInterface) {
            $task->setParameters($parameters);
        }

        $task->execute();

        $stateStorage->persist($parameterBagId, $parameters);
    }
232085
public function addProjectToTask($taskId, $projectId = null)
    {
        $data = [
            'project' => $projectId ?: $this->defaultProjectId
        ];

        return $this->curl->post("tasks/{$taskId}/addProject", ['data' => $data]);
    }
134575
public function onWorkerStart($serv, $workerId)
    {
        $this->initAsynPools();
        $this->initRedisProxies();
        //注册
        $this->asynPoolManager = new AsynPoolManager(null, $this);
        $this->asynPoolManager->noEventAdd();
        foreach ($this->asynPools as $pool) {
            if ($pool) {
                $pool->workerInit($workerId);
                $this->asynPoolManager->registerAsyn($pool);
            }
        }

        //redis proxy监测
        getInstance()->sysTimers[] = swoole_timer_tick(5000, function () {
            if (!empty($this->redisProxyManager)) {
                foreach ($this->redisProxyManager as $proxy) {
                    $proxy->check();
                }
            }
        });
    }
263181
public static function send(string $data, $fd, $from_id = 0) {
		self::$server->send($fd, $data, $from_id);
	}
1334560
public static function workerStartApp($server, $target)
    {
        if(!static::$workerStartAppComplete)
        {
            // 触发项目的workerstart事件
            Event::trigger('IMI.MAIN_SERVER.WORKER.START.APP', [
                'server'    => $server,
                'workerID'  => static::$workerID,
            ], $target, WorkerStartEventParam::class);
            static::$workerStartAppComplete = true;
        }
    }