diff --git a/kiri-task/AsyncTaskExecute.php b/kiri-task/AsyncTaskExecute.php index ba911254..ec067ceb 100644 --- a/kiri-task/AsyncTaskExecute.php +++ b/kiri-task/AsyncTaskExecute.php @@ -8,7 +8,6 @@ use Kiri\Abstracts\Component; use Kiri\Server\SwooleServerInterface; use Psr\Container\ContainerExceptionInterface; use Psr\Container\NotFoundExceptionInterface; -use Swoole\Process; /** @@ -18,35 +17,17 @@ class AsyncTaskExecute extends Component { - use TaskResolve; - - - private int $total = 50; - - /** - * @param int $total - */ - public function setTotal(int $total): void - { - $this->total = $total; - } + public TaskManager $hashMap; /** * @return void - * @throws Kiri\Exception\ConfigException * @throws ContainerExceptionInterface * @throws NotFoundExceptionInterface */ - public function start() + public function init() { - $processManager = $this->getContainer()->get(Kiri\Server\ProcessManager::class); - for ($i = 0; $i < $this->total; $i++) { - $class = new TaskProcess(); - $class->name = 'task.' . $i; - - $processManager->add($class, 'tasker'); - } + $this->hashMap = $this->getContainer()->get(TaskManager::class); } @@ -56,7 +37,6 @@ class AsyncTaskExecute extends Component * @param int $workerId * @throws ContainerExceptionInterface * @throws NotFoundExceptionInterface - * @throws \ReflectionException * @throws Exception */ public function execute(OnTaskInterface|string $handler, array $params = [], int $workerId = -1) @@ -71,18 +51,29 @@ class AsyncTaskExecute extends Component $workerId = random_int(0, $server->setting['task_worker_num'] - 1); } $server->task(serialize($handler), $workerId); - } else { - if ($workerId < 0 || $workerId > $this->total) { - $workerId = random_int(0, $this->total - 1); - } - - $processManager = $container->get(Kiri\Server\ProcessManager::class); - - /** @var Process $process */ - $process = $processManager->get('task.' . $workerId, 'tasker'); - $process->write(serialize($handler)); } } + /** + * @param $handler + * @param $params + * @return object + * @throws ContainerExceptionInterface + * @throws NotFoundExceptionInterface + * @throws \ReflectionException + * @throws Exception + */ + private function handle($handler, $params): object + { + if (!class_exists($handler) && $this->hashMap->has($handler)) { + $handler = $this->hashMap->get($handler); + } + $implements = $this->getContainer()->getReflect($handler); + if (!in_array(OnTaskInterface::class, $implements->getInterfaceNames())) { + throw new Exception('Task must instance ' . OnTaskInterface::class); + } + return $implements->newInstanceArgs($params); + } + } diff --git a/kiri-task/CoroutineTaskExecute.php b/kiri-task/CoroutineTaskExecute.php deleted file mode 100644 index 246605ba..00000000 --- a/kiri-task/CoroutineTaskExecute.php +++ /dev/null @@ -1,95 +0,0 @@ -total = $total; - } - - - /** - * - */ - public function start() - { - $this->hashMap = new HashMap(); - - $this->channel = new Coroutine\Channel($this->total); - - $this->taskServer = \Kiri::getDi()->get(OnServerTask::class); - - Coroutine::create(function () { - $barrier = Coroutine\Barrier::make(); - for ($i = 0; $i < $this->total; $i++) { - Coroutine::create(function () { - $this->handler(); - }); - } - Coroutine\Barrier::wait($barrier); - }); - } - - - /** - * @return void - * @throws ConfigException - */ - protected function handler() - { - Coroutine\defer(function () { - $this->handler(); - }); - $data = $this->channel->pop(-1); - - $task = new Task(); - $task->data = $data; - - $this->taskServer->onCoroutineTask(null, $task); - } - - - /** - * @param OnTaskInterface|string $handler - * @param array $params - * @param int $workerId - * @return void - * @throws ReflectionException - */ - public function execute(OnTaskInterface|string $handler, array $params = [], int $workerId = -1) - { - if (is_string($handler)) { - $handler = $this->handle($handler, $params); - } - $this->channel->push(serialize($handler)); - } - -} diff --git a/kiri-task/TaskManager.php b/kiri-task/TaskManager.php index 2b6be615..9a1a9e74 100644 --- a/kiri-task/TaskManager.php +++ b/kiri-task/TaskManager.php @@ -2,6 +2,7 @@ namespace Kiri\Task; +use JetBrains\PhpStorm\Pure; use Kiri\Abstracts\Component; use Kiri\Core\HashMap; use Psr\Container\ContainerExceptionInterface; @@ -58,6 +59,16 @@ class TaskManager extends Component } + /** + * @param string $key + * @return bool + */ + #[Pure] public function has(string $key): bool + { + return $this->hashMap->has($key); + } + + /** * @param string $key * @return OnTaskInterface @@ -69,6 +80,9 @@ class TaskManager extends Component $task = $this->hashMap->get($key); if (is_string($task)) { $task = $this->getContainer()->get($task); + if (!empty($task)) { + $this->add($key, $task); + } } return $task; } diff --git a/kiri-task/TaskProcess.php b/kiri-task/TaskProcess.php deleted file mode 100644 index 376d2f22..00000000 --- a/kiri-task/TaskProcess.php +++ /dev/null @@ -1,63 +0,0 @@ -index = $index; - } - - - /** - * @return string - */ - public function getName(): string - { - return 'task.' . $this->index; - } - - - /** - * @param Process $process - * @throws \Exception - */ - public function process(Process $process): void - { - $task = \Kiri::getContainer()->get(OnServerTask::class); - while (!$this->isStop()) { - $read = $process->read(); - - $task->onTask(null, 0, 0, $read); - } - } - - - /** - * @return $this - */ - public function onSigterm(): static - { - pcntl_signal(SIGTERM, function () { - $this->isStop = true; - }); - return $this; - } - - -} diff --git a/kiri-task/TaskResolve.php b/kiri-task/TaskResolve.php deleted file mode 100644 index 8b1401c7..00000000 --- a/kiri-task/TaskResolve.php +++ /dev/null @@ -1,31 +0,0 @@ -hashMap->has($handler)) { - $handler = $this->hashMap->get($handler); - } - $implements = $this->getContainer()->getReflect($handler); - if (!in_array(OnTaskInterface::class, $implements->getInterfaceNames())) { - throw new Exception('Task must instance ' . OnTaskInterface::class); - } - return $implements->newInstanceArgs($params); - } - -}