diff --git a/System/Crontab.php b/System/Crontab.php index d17ec4e4..fabb82b7 100644 --- a/System/Crontab.php +++ b/System/Crontab.php @@ -41,6 +41,16 @@ class Crontab extends BaseObject private int $execute_number = 0; + + /** + * @return $this + */ + public function increment(): static + { + $this->execute_number += 1; + return $this; + } + /** * @return array|Closure */ @@ -70,7 +80,7 @@ class Crontab extends BaseObject */ public function getTickTime(): int { - return $this->tickTime; + return $this->tickTime * 1000; } /** @@ -186,18 +196,10 @@ class Crontab extends BaseObject /** * @throws Exception */ - public function execute(CrontabProcess $process): void + public function execute(): void { try { - $this->warning('execute crontab.'); - call_user_func($this->handler, $this->params); - $this->execute_number += 1; - if ($this->execute_number >= $this->max_execute_number) { - $process->clear($this->getName()); - } else if (!$this->isLoop()) { - $process->clear($this->getName()); - } } catch (\Throwable $throwable) { $this->addError($throwable->getMessage()); } finally { diff --git a/System/Process/CrontabProcess.php b/System/Process/CrontabProcess.php index b2297ffc..e5f585d3 100644 --- a/System/Process/CrontabProcess.php +++ b/System/Process/CrontabProcess.php @@ -26,28 +26,75 @@ class CrontabProcess extends Process public array $names = []; + public array $scores = []; + public array $timers = []; + + /** * @param \Swoole\Process $process */ public function onHandler(\Swoole\Process $process): void { - try { - $content = $process->read(); + $this->readBySocket($process); + Timer::tick(1000, function () { + $startTime = time(); - $_content = json_decode($content, true); - if (is_null($_content)) { - $this->jobDelivery($content); - } else { - $this->otherAction($_content); + $time = $this->scores[$startTime]; + unset($this->scores[$startTime]); + foreach ($time as $value) { + Coroutine::create(function (Crontab $value, int $startTime) { + $this->dispatch($value, $startTime); + }, $value, $startTime); + } + }); + } + + + /** + * @param Crontab $value + * @param int $startTime + * @throws \Exception + */ + private function dispatch(Crontab $value, int $startTime) + { + try { + $value->increment()->execute(); + if ($value->getExecuteNumber() < $value->getMaxExecuteNumber()) { + $this->addTask($value, $startTime + $value->getTickTime()); + } else if ($value->isLoop()) { + $this->addTask($value, $startTime + $value->getTickTime()); } } catch (\Throwable $exception) { $this->application->error($exception->getMessage()); - } finally { - $this->onHandler($process); } } + /** + * @param \Swoole\Process $process + * @throws \Exception + */ + public function readBySocket(\Swoole\Process $process) + { + Coroutine::create(function (\Swoole\Process $process) { + try { + $content = $process->read(); + + $_content = json_decode($content, true); + if (is_null($_content)) { + $this->jobDelivery($content); + } else { + $this->otherAction($_content); + } + } catch (\Throwable $exception) { + $this->application->error($exception->getMessage()); + } finally { + $this->onHandler($process); + } + }, $process); + } + + /** * @param $content */ @@ -76,8 +123,13 @@ class CrontabProcess extends Process if (!isset($this->names[$name])) { return; } - Timer::exists($this->names[$name]) && Timer::clear($this->names[$name]); - unset($this->names[$name]); + $timers = $this->timers[$name]; + + $search = array_search($name, $this->scores[$timers]); + if ($search !== false) { + unset($this->scores[$timers][$search]); + } + unset($this->timers[$name], $this->names[$name]); } @@ -89,22 +141,35 @@ class CrontabProcess extends Process /** @var Crontab $content */ $content = unserialize($content); + $ticker = intval($content->getTickTime() * 1000) + time(); + + $this->addTask($content, $ticker); + } + + + /** + * @param Crontab $content + * @param $ticker + */ + private function addTask(Crontab $content, $ticker) + { $name = $content->getName(); if (isset($this->names[$name])) { - Timer::clear($this->names[$name]); + unset($this->names[$content->getName()]); + + $search = array_search($content->getName(), $this->scores); + unset($this->scores[$search]); } - if ($content->isLoop()) { - $this->names[$name] = Timer::tick(intval($content->getTickTime() * 1000), function ($content) { - var_dump(Timer::class); -// $content->execute($this); - }, $content); - } else { - $this->names[$name] = Timer::after(intval($content->getTickTime() * 1000), function ($content) { - var_dump(Timer::class); -// $content->execute($this); - }, $content); + + if (!isset($this->scores[$ticker])) { + $this->scores[$ticker] = []; } - var_dump($this->names); + + $this->timers[$content->getName()] = $ticker; + $this->scores[$ticker][] = $content->getName(); + $this->names[$content->getName()] = $content; + + ksort($this->scores, SORT_NUMERIC); }