Swoole 异步任务结束事件

@param  \Swoole\Server   $server
@param  int              $taskId
@param  mixed            $data

@throws \RuntimeException
326293
public function onFinish(Server $server, int $taskId, $data)
    {
        array_walk($this->handlers[$data['name']], function ($name) {
            $handler = $this->createHandler($name);
            $handler->finish($data['result']);
            unset($handler);
        });
    }
77986
public void updateText(int taskId, String text) {
		getResourceFactory().getApiResource("/task/" + taskId + "/text")
				.entity(new TaskText(text), MediaType.APPLICATION_JSON_TYPE)
				.put();
	}
798282
public function onSwooleWorkerStart($serv, $workerId)
    {
        parent::onSwooleWorkerStart($serv, $workerId);
        $this->initAsynPools($workerId);
        //进程锁保证只有一个进程会执行以下的代码,reload也不会执行
        if (!$this->isTaskWorker() && $this->initLock->trylock()) {
            //进程启动后进行开服的初始化
            $this->onOpenServiceInitialization();
            if (Start::$testUnity) {
                $ready = ProcessManager::getInstance()->getRpcCall(CatCacheProcess::class)->isReady();
                if (!$ready) {
                    EventDispatcher::getInstance()->addOnceCoroutine(CatCacheProcess::READY);
                }
                new TestModule(Start::$testUnityDir);
            }
            $this->initLock->lock_read();
        }
        //向SDHelp進程取數據
        if (!$this->isTaskWorker()) {
            $isReload = $this->isReload();
            ConsulHelp::start();
            TimerTask::start();
            if ($this->config->get('catCache.enable', false)) {
                TimerCallBack::init();
                if (!$isReload) {
                    $ready = ProcessManager::getInstance()->getRpcCall(CatCacheProcess::class)->isReady();
                    if (!$ready) {
                        EventDispatcher::getInstance()->addOnceCoroutine(CatCacheProcess::READY);
                    }
                }
                Actor::recovery($workerId);
            }
        }
        //Code coverage
        register_tick_function([$this, 'onPhpTick']);
    }
72061
public function onWorkerStart($server)
    {
        if ($this->container->config->get('swoole.enable_coroutine', false)) {
            Runtime::enableCoroutine(true);
        }

        $this->clearCache();

        $this->container->event->trigger('swoole.workerStart', func_get_args());

        // don't init app in task workers
        if ($server->taskworker) {
            $this->setProcessName('task process');

            return;
        }

        $this->setProcessName('worker process');

        $this->prepareApplication();

        if ($this->isServerWebsocket) {
            $this->prepareWebsocketHandler();
            $this->loadWebsocketRoutes();
        }
    }
511912
public function onWorkerError(SwServer $server, int $workerId, int $workerPid, int $exitCode)
    {
        $workerType = $server->taskworker ? 'task' : 'worker';

        // {{ log
        $this->get('logger')->info('Worker 进程异常退出', [
            'id' => $workerId,
            'pid' => $workerPid,
            'type' => ucfirst($workerType),
            'exitcode' => $exitCode,
        ]);
        // }}

        unset($workerType);

        $this->callProtocolMethod('onError', $server, $workerId);
    }
611489
public function exists(string $taskID, string $listID = ''): bool
    {
        $list = $this->getListData($listID);
        if (isset($list[$taskID])) {
            $app = App::get();
            return $app->data->exists($this->getTaskDataKey($taskID, $listID));
        }
        return false;
    }
920791
public function onWorkerStop(swoole_server $server, $worker_id)
    {
        $this->console(sprintf('Server <info>%s</info> Worker[<info>%s</info>] is shutdown', $this->name, $worker_id), OutputInterface::VERBOSITY_DEBUG);
    }
795152
public static function onTask($serv, $task_id, $worker_id, $data) {
		$rs = Plugin::trigger('taskStart', [$task_id, $worker_id, $data]);
		if (is_string($rs)) {
			return $rs;
		}
	}
1089240
public static function nPostWait(string $name, $data, $timeout, $workerID = -1)
    {
        $task = TaskParser::getInstance()->getTask($name);
        $paramClass = $task['Task']->paramClass;
        return static::nPostWait(new TaskInfo(new $task['className'], new $paramClass($data)), $timeout, $workerID);
    }
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);
    }
779539
public function onWorkerStart(swoole_server $server, $worker_id)
    {
        process_rename($this->getName() . ' worker');

        $this->console(sprintf('Server Worker[<info>%s</info>] is started [<info>%s</info>]', $server->worker_pid, $worker_id), OutputInterface::VERBOSITY_DEBUG);
    }
718305
public function onWorkerStop(Server $server,int $worker_id)
    {
        $this->fireServerEvent(SwooleEvent::ON_WORKER_STOP, [$server,$worker_id]);
    }
735988
public function handle(AbstractServer $server): void
    {
        parent::handle($server); // TODO: Change the autogenerated stub

        /* @var TaskContract $object */
        $object = $this->data['object'];
        /* @var array $params */
        $params = $this->data['params'];

        try {
            $result = $object->handle($server, ...$params);

            $this->server->finish($result);
        } catch (\Throwable $exception) {
            if (method_exists($object, 'failed')) {
                $object->failed($exception);
            }

            throw $exception;
        }
    }
1337349
private function beforeTask(
        string $logid,
        int $spanid,
        string $name,
        string $method,
        string $type,
        string $taskClass
    ) {
        $event = new BeforeTaskEvent(TaskEvent::BEFORE_TASK, $logid, $spanid, $name, $method, $type, $taskClass);
        App::trigger($event);
    }
932859
public function doWork(swoole_server $server, $fd, $data, $from_id)
    {
        switch (trim($data)) {
            case 'quit':
                $server->send($fd, 'connection closed');
                $server->close($fd);

                break;
            case 'reload':
                $this->getSwoole()->reload();

                break;
            case 'status':
            default:
                $info = $server->stats();
                $status = '';
                foreach ($info as $key => $value) {
                    $status .= '['.date('Y-m-d H:i:s').']: '.$key.': '.$value.PHP_EOL;
                }
                $server->send($fd, $status);

                break;
        }
    }
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();
                }
            }
        });
    }
829478
public function setSwoole(Server $swoole)
    {
        $enabled = isset($swoole->setting['task_worker_num']) &&
                    (int) $swoole->setting['task_worker_num'] > 0;

        if ($enable) {
            $this->swoole = $swoole;
        }

        $this->setEnable($enabled);

        unset($enabled);
    }