diff --git a/System/Crontab/Zookeeper.php b/System/Crontab/Zookeeper.php index 5ce89352..18ad15f5 100644 --- a/System/Crontab/Zookeeper.php +++ b/System/Crontab/Zookeeper.php @@ -5,13 +5,12 @@ namespace Snowflake\Crontab; use Exception; -use HttpServer\Server; use Snowflake\Abstracts\Config; use Snowflake\Cache\Redis; use Snowflake\Exception\ConfigException; use Snowflake\Process\Process; use Snowflake\Snowflake; -use Swoole\Coroutine; +use Swoole\Timer; use Throwable; /** @@ -22,57 +21,65 @@ class Zookeeper extends Process { - private int $workerNum = 0; + private int $workerNum = 0; - private mixed $server; + private mixed $server; /** * @return string * @throws ConfigException */ - public function getProcessName(): string - { - $name = Config::get('id', 'system') . '[' . $this->pid . ']'; - if (!empty($prefix)) { - $name .= '.Crontab zookeeper'; - } - return $name; - } + public function getProcessName(): string + { + $name = Config::get('id', 'system') . '[' . $this->pid . ']'; + if (!empty($prefix)) { + $name .= '.Crontab zookeeper'; + } + return $name; + } - /** - * @param \Swoole\Process $process - * @throws Exception - */ - public function before(\Swoole\Process $process): void - { - /** @var Producer $crontab */ - $crontab = Snowflake::app()->get('crontab'); - $crontab->clearAll(); + /** + * @param \Swoole\Process $process + * @throws Exception + */ + public function before(\Swoole\Process $process): void + { + /** @var Producer $crontab */ + $crontab = Snowflake::app()->get('crontab'); + $crontab->clearAll(); - $this->server = $server = Snowflake::app()->getSwoole(); - $this->workerNum = $server->setting['worker_num'] + $server->setting['task_worker_num']; - } + $this->server = $server = Snowflake::app()->getSwoole(); + $this->workerNum = $server->setting['worker_num'] + $server->setting['task_worker_num']; + } - /** - * @param \Swoole\Process $process - * @throws Exception - */ - public function onHandler(\Swoole\Process $process): void - { - $redis = Snowflake::app()->getRedis(); - while (true) { - $range = $this->loadCarobTask($redis); - var_dump($range); - foreach ($range as $value) { - $this->dispatch($redis, $value); - } - Coroutine::sleep(80); - } - } + /** + * @param \Swoole\Process $process + * @throws Exception + */ + public function onHandler(\Swoole\Process $process): void + { + Timer::tick(80, [$this, 'loop']); + } + + + /** + * @throws ConfigException + * @throws Exception + */ + public function loop() + { + $redis = Snowflake::app()->getRedis(); + defer(fn() => $redis->release()); + $range = $this->loadCarobTask($redis); + var_dump($range); + foreach ($range as $value) { + $this->dispatch($redis, $value); + } + } /** @@ -80,43 +87,43 @@ class Zookeeper extends Process * @param $value * @throws Exception */ - private function dispatch(Redis|\Redis $redis, $value) - { - try { - $params['action'] = 'crontab'; - if (empty($handler = $redis->get('crontab:' . $value))) { - return; - } - $params['handler'] = swoole_unserialize($handler); + private function dispatch(Redis|\Redis $redis, $value) + { + try { + $params['action'] = 'crontab'; + if (empty($handler = $redis->get('crontab:' . $value))) { + return; + } + $params['handler'] = swoole_unserialize($handler); - $this->server->sendMessage($params, $this->getWorker()); - } catch (Throwable $exception) { - logger()->addError($exception); - } - } + $this->server->sendMessage($params, $this->getWorker()); + } catch (Throwable $exception) { + logger()->addError($exception); + } + } - /** - * @return int - * @throws \Exception - */ - private function getWorker(): int - { - return random_int(0, $this->workerNum - 1); - } + /** + * @return int + * @throws \Exception + */ + private function getWorker(): int + { + return random_int(0, $this->workerNum - 1); + } /** * @param Redis|\Redis $redis * @return array */ - private function loadCarobTask(Redis|\Redis $redis): array - { - $range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)time()); + private function loadCarobTask(Redis|\Redis $redis): array + { + $range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)time()); - $redis->zRem(Producer::CRONTAB_KEY, ...$range); + $redis->zRem(Producer::CRONTAB_KEY, ...$range); - return $range; - } + return $range; + } }