diff --git a/System/Abstracts/Crontab.php b/System/Abstracts/Crontab.php index 8fca60f4..d366cd8e 100644 --- a/System/Abstracts/Crontab.php +++ b/System/Abstracts/Crontab.php @@ -4,6 +4,8 @@ namespace Snowflake\Abstracts; +use Snowflake\Core\Json; +use Snowflake\Process\CrontabProcess; use Snowflake\Snowflake; use Exception; @@ -16,51 +18,40 @@ class Crontab extends Component { - /** - * @param \Snowflake\Crontab $crontab - * @param $executeTime - * @throws Exception - */ - public function dispatch(\Snowflake\Crontab $crontab, int $executeTime) - { - $redis = Snowflake::app()->getRedis(); - - $redis->zAdd('system:crontab', (string)$executeTime, serialize($crontab)); - } + /** + * @param \Snowflake\Crontab $crontab + * @param $executeTime + * @throws Exception + */ + public function dispatch(\Snowflake\Crontab $crontab) + { + /** @var CrontabProcess $redis */ + $redis = Snowflake::app()->get(CrontabProcess::class); + $redis->write(serialize($crontab)); + } - /** - * @param string $name - * @throws Exception - */ - public function clear(string $name) - { - $redis = Snowflake::app()->getRedis(); - - $data = $redis->zRevRange('system:crontab', 0, -1); - if (empty($data)) { - return; - } - foreach ($data as $datum) { - /** @var \Snowflake\Crontab $crontab */ - $crontab = unserialize($datum); - if ($crontab->getName() == $name) { - $redis->zRem('system:crontab', $datum); - } - } - - } + /** + * @param string $name + * @throws Exception + */ + public function clear(string $name) + { + /** @var CrontabProcess $redis */ + $redis = Snowflake::app()->get(CrontabProcess::class); + $redis->write(Json::encode(['action' => 'clear', 'name' => $name])); + } - /** - * @throws Exception - */ - public function clearAll() - { - $redis = Snowflake::app()->getRedis(); - - $redis->del('system:crontab'); - } + /** + * @throws Exception + */ + public function clearAll() + { + /** @var CrontabProcess $redis */ + $redis = Snowflake::app()->get(CrontabProcess::class); + $redis->write(Json::encode(['action' => 'clearAll'])); + } } diff --git a/System/Crontab.php b/System/Crontab.php index 36cbe9dc..b3c2e756 100644 --- a/System/Crontab.php +++ b/System/Crontab.php @@ -7,6 +7,7 @@ namespace Snowflake; use Closure; use Exception; use Snowflake\Abstracts\BaseObject; +use Swoole\Timer; /** * Class Async @@ -16,163 +17,189 @@ class Crontab extends BaseObject { - private array|Closure $handler; + private array|Closure $handler; - private string $name = ''; + private string $name = ''; - private mixed $params; + private mixed $params; - private int $tickTime = 1; + private int $tickTime = 1; - private bool $isLoop = false; + private bool $isLoop = false; - private int $max_execute_number = -1; + private int $timerId = -1; - private int $execute_number = 0; - - /** - * @return array|Closure - */ - public function getHandler(): array|Closure - { - return $this->handler; - } - - /** - * @return string - */ - public function getName(): string - { - return $this->name; - } - - /** - * @return mixed - */ - public function getParams(): mixed - { - return $this->params; - } - - /** - * @return int - */ - public function getTickTime(): int - { - return $this->tickTime; - } - - /** - * @return bool - */ - public function isLoop(): bool - { - return $this->isLoop; - } - - /** - * @return int - */ - public function getMaxExecuteNumber(): int - { - return $this->max_execute_number; - } - - /** - * @return int - */ - public function getExecuteNumber(): int - { - return $this->execute_number; - } + private int $max_execute_number = -1; - /** - * @param array|Closure $handler - */ - public function setHandler(array|Closure $handler): void - { - $this->handler = $handler; - } + private int $execute_number = 0; - /** - * @param string $name - */ - public function setName(string $name): void - { - $this->name = $name; - } + /** + * @return array|Closure + */ + public function getHandler(): array|Closure + { + return $this->handler; + } - /** - * @param mixed $params - */ - public function setParams(mixed $params): void - { - $this->params = $params; - } + /** + * @return string + */ + public function getName(): string + { + return $this->name; + } - /** - * @param int $tickTime - */ - public function setTickTime(int $tickTime): void - { - $this->tickTime = $tickTime; - } + /** + * @return mixed + */ + public function getParams(): mixed + { + return $this->params; + } - /** - * @param bool $isLoop - */ - public function setIsLoop(bool $isLoop): void - { - $this->isLoop = $isLoop; - } + /** + * @return int + */ + public function getTickTime(): int + { + return $this->tickTime; + } - /** - * @param int $max_execute_number - */ - public function setMaxExecuteNumber(int $max_execute_number): void - { - $this->max_execute_number = $max_execute_number; - } + /** + * @return bool + */ + public function isLoop(): bool + { + return $this->isLoop; + } - /** - * @param int $execute_number - */ - public function setExecuteNumber(int $execute_number): void - { - $this->execute_number = $execute_number; - } + /** + * @return int + */ + public function getMaxExecuteNumber(): int + { + return $this->max_execute_number; + } + + /** + * @return int + */ + public function getExecuteNumber(): int + { + return $this->execute_number; + } - /** - * @throws Exception - */ - public function execute(): void - { - $redis = Snowflake::app()->getRedis(); - try { - $this->execute_number += 1; - call_user_func($this->handler, $this->params); - if ($this->isLoop === false) { - return; - } - if ($this->max_execute_number === -1) { - $redis->zAdd('system:crontab', time() + $this->tickTime, serialize($this)); - } else if ($this->execute_number < $this->max_execute_number) { - $redis->zAdd('system:crontab', time() + $this->tickTime, serialize($this)); - } - } catch (\Throwable $throwable) { - $this->addError($throwable->getMessage()); - } finally { - $redis->release(); - } - } + /** + * @param array|Closure $handler + */ + public function setHandler(array|Closure $handler): void + { + $this->handler = $handler; + } + + /** + * @param string $name + */ + public function setName(string $name): void + { + $this->name = $name; + } + + /** + * @param mixed $params + */ + public function setParams(mixed $params): void + { + $this->params = $params; + } + + /** + * @param int $tickTime + */ + public function setTickTime(int $tickTime): void + { + $this->tickTime = $tickTime; + } + + /** + * @param bool $isLoop + */ + public function setIsLoop(bool $isLoop): void + { + $this->isLoop = $isLoop; + } + + /** + * @param int $max_execute_number + */ + public function setMaxExecuteNumber(int $max_execute_number): void + { + $this->max_execute_number = $max_execute_number; + } + + /** + * @param int $execute_number + */ + public function setExecuteNumber(int $execute_number): void + { + $this->execute_number = $execute_number; + } + + /** + * @return int + */ + public function getTimerId(): int + { + return $this->timerId; + } + + /** + * @param int $timerId + */ + public function setTimerId(int $timerId): void + { + $this->timerId = $timerId; + } + + + /** + * + */ + public function clearTimer() + { + $this->warning('crontab timer clear.'); + if (Timer::exists($this->timerId)) { + Timer::clear($this->timerId); + } + } + + + /** + * @throws Exception + */ + public function execute(): void + { + try { + call_user_func($this->handler, $this->params); + $this->execute_number += 1; + + if ($this->execute_number < $this->max_execute_number) { + $this->clearTimer(); + } + } catch (\Throwable $throwable) { + $this->addError($throwable->getMessage()); + } finally { + fire(Event::SYSTEM_RESOURCE_RELEASES); + } + } } diff --git a/System/Process/CrontabProcess.php b/System/Process/CrontabProcess.php index 6fe14a2d..897238cb 100644 --- a/System/Process/CrontabProcess.php +++ b/System/Process/CrontabProcess.php @@ -5,6 +5,7 @@ namespace Snowflake\Process; use ReflectionException; +use Snowflake\Core\Json; use Snowflake\Crontab; use Snowflake\Exception\ComponentException; use Snowflake\Exception\ConfigException; @@ -23,67 +24,70 @@ use Swoole\Timer; class CrontabProcess extends Process { - - public Channel $channel; + /** @var Crontab[] $names */ + public array $names = []; - /** - * @param \Swoole\Process $process - */ - public function onHandler(\Swoole\Process $process): void - { - $this->channel = new Channel(5000); - - Coroutine::create([$this, 'execute']); - - Timer::tick(1000, [$this, 'systemLoop']); - } + /** + * @param \Swoole\Process $process + */ + public function onHandler(\Swoole\Process $process): void + { + while (true) { + $content = $process->read(); + $_content = json_decode($content, true); + if (is_null($_content)) { + $this->jobDelivery($content); + } else { + $this->otherAction($content); + } + } + } - /** - * @throws \Exception - */ - public function execute() - { - while (true) { - /** @var Crontab $list */ - $list = $this->channel->pop(-1); - $list->execute(); - } - } + /** + * @param $content + */ + private function otherAction($content) + { + call_user_func(match ($content['action']) { + 'clear' => function ($content) { + if (!isset($this->names[$content['name']])) { + return; + } + $this->names[$content['name']]->clearTimer(); + }, + 'clearAll' => function () { + foreach ($this->names as $name => $crontab) { + $crontab->clearTimer(); + } + }, + default => function () { + $this->application->error('unknown action'); + } + }); + } - /** - * @throws ReflectionException - * @throws ComponentException - * @throws ConfigException - * @throws NotFindClassException - * @throws Exception - * @throws \Exception - */ - public function systemLoop() - { - $score = time(); - $redis = Snowflake::app()->getRedis(); + /** + * @param $content + */ + private function jobDelivery($content) + { + $content = unserialize($content); + $this->names[$content->getName()] = $content; + if (!($content instanceof Crontab)) { + return; + } + $runTicker = [$content, 'execute']; + $timer = $content->getTickTime() * 1000; + if ($content->isLoop()) { + $timerId = Timer::tick($timer, $runTicker); + } else { + $timerId = Timer::after($timer, $runTicker); + } + $content->setTimerId($timerId); + } - $lists = $redis->zRangeByScore('system:crontab', '0', (string)$score); - $redis->zRemRangeByScore('system:crontab', '0', (string)$score); - - if (empty($lists)) { - $redis->release(); - return; - } - - $barrier = Barrier::make(); - foreach ($lists as $list) { - $list = unserialize($list); - if (!($list instanceof Crontab)) { - continue; - } - $this->channel->push($list); - } - Barrier::wait($barrier); - $redis->release(); - } }