diff --git a/kiri-task/AsyncTaskExecute.php b/kiri-task/AsyncTaskExecute.php index c9ab5924..ba911254 100644 --- a/kiri-task/AsyncTaskExecute.php +++ b/kiri-task/AsyncTaskExecute.php @@ -6,7 +6,9 @@ use Exception; use Kiri; use Kiri\Abstracts\Component; use Kiri\Server\SwooleServerInterface; -use Swoole\Coroutine; +use Psr\Container\ContainerExceptionInterface; +use Psr\Container\NotFoundExceptionInterface; +use Swoole\Process; /** @@ -19,10 +21,42 @@ class AsyncTaskExecute extends Component use TaskResolve; + private int $total = 50; + + /** + * @param int $total + */ + public function setTotal(int $total): void + { + $this->total = $total; + } + + + /** + * @return void + * @throws Kiri\Exception\ConfigException + * @throws ContainerExceptionInterface + * @throws NotFoundExceptionInterface + */ + public function start() + { + $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'); + } + } + + /** * @param OnTaskInterface|string $handler * @param array $params * @param int $workerId + * @throws ContainerExceptionInterface + * @throws NotFoundExceptionInterface + * @throws \ReflectionException * @throws Exception */ public function execute(OnTaskInterface|string $handler, array $params = [], int $workerId = -1) @@ -30,12 +64,24 @@ class AsyncTaskExecute extends Component if (is_string($handler)) { $handler = $this->handle($handler, $params); } + $container = $this->getContainer(); + if ($container->has(SwooleServerInterface::class)) { + $server = $container->get(SwooleServerInterface::class); + if ($workerId < 0 || $workerId > $server->setting['task_worker_num']) { + $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); + } - $server = Kiri::getDi()->get(SwooleServerInterface::class); - if ($workerId < 0 || $workerId > $server->setting['task_worker_num']) { - $workerId = random_int(0, $server->setting['task_worker_num'] - 1); + $processManager = $container->get(Kiri\Server\ProcessManager::class); + + /** @var Process $process */ + $process = $processManager->get('task.' . $workerId, 'tasker'); + $process->write(serialize($handler)); } - $server->task(serialize($handler), $workerId); } diff --git a/kiri-task/CoroutineTaskExecute.php b/kiri-task/CoroutineTaskExecute.php index cf6f66e0..246605ba 100644 --- a/kiri-task/CoroutineTaskExecute.php +++ b/kiri-task/CoroutineTaskExecute.php @@ -27,10 +27,19 @@ class CoroutineTaskExecute extends Component private int $total = 50; + /** + * @param int $total + */ + public function setTotal(int $total): void + { + $this->total = $total; + } + + /** * */ - public function init() + public function start() { $this->hashMap = new HashMap(); @@ -40,7 +49,7 @@ class CoroutineTaskExecute extends Component Coroutine::create(function () { $barrier = Coroutine\Barrier::make(); - for ($i = 0; $i < 50; $i++) { + for ($i = 0; $i < $this->total; $i++) { Coroutine::create(function () { $this->handler(); }); diff --git a/kiri-task/TaskProcess.php b/kiri-task/TaskProcess.php new file mode 100644 index 00000000..376d2f22 --- /dev/null +++ b/kiri-task/TaskProcess.php @@ -0,0 +1,63 @@ +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; + } + + +}