From bd57ba367ac1c50508604c4a2afb715e5768220e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mr=C2=B7x?= Date: Thu, 15 Apr 2021 18:18:41 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E5=90=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- HttpServer/Events/OnPipeMessage.php | 69 ++++++++------- System/Crontab/Zookeeper.php | 129 +++++++++++++++------------- 2 files changed, 103 insertions(+), 95 deletions(-) diff --git a/HttpServer/Events/OnPipeMessage.php b/HttpServer/Events/OnPipeMessage.php index 68fa891e..ee738681 100644 --- a/HttpServer/Events/OnPipeMessage.php +++ b/HttpServer/Events/OnPipeMessage.php @@ -9,7 +9,6 @@ use HttpServer\Abstracts\Callback; use Kafka\Struct; use Snowflake\Crontab\Crontab; use Snowflake\Event; -use Snowflake\Exception\ComponentException; use Swoole\Server; /** @@ -25,20 +24,20 @@ class OnPipeMessage extends Callback * @param $swollen_universalize * @throws Exception */ - public function onHandler(Server $server, int $src_worker_id, $swollen_universalize) - { - try { - match ($swollen_universalize['action'] ?? null) { - 'kafka' => $this->onKafkaWorker($swollen_universalize), - 'crontab' => $this->onCrontabWorker($swollen_universalize), - default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize) - }; - } catch (\Throwable $exception) { - $this->addError($exception); - } finally { - fire(Event::SYSTEM_RESOURCE_RELEASES); - } - } + public function onHandler(Server $server, int $src_worker_id, $swollen_universalize) + { + try { + match ($swollen_universalize['action'] ?? null) { + 'kafka' => $this->onKafkaWorker($swollen_universalize), + 'crontab' => $this->onCrontabWorker($swollen_universalize), + default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize) + }; + } catch (\Throwable $exception) { + $this->addError($exception); + } finally { + fire(Event::SYSTEM_RESOURCE_RELEASES); + } + } /** @@ -46,13 +45,17 @@ class OnPipeMessage extends Callback * @return string * @throws Exception */ - private function onCrontabWorker(array $message): string - { - /** @var Crontab $crontab */ - $crontab = $message['handler'] ?? null; - $crontab->increment()->execute(); - return 'success'; - } + private function onCrontabWorker(array $message): string + { + /** @var Crontab $crontab */ + $crontab = $message['handler'] ?? null; + if (!isset($message['handler'])) { + throw new Exception('unknown handler'); + } + $crontab = swoole_unserialize($crontab['handler']); + $crontab->increment()->execute(); + return 'success'; + } /** @@ -62,26 +65,26 @@ class OnPipeMessage extends Callback * @return string * @throws Exception */ - private function onMessageWorker($server, $src_worker_id, $message): string - { - fire(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]); + private function onMessageWorker($server, $src_worker_id, $message): string + { + fire(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]); - return 'success'; - } + return 'success'; + } /** * @param array $message * @return string */ - private function onKafkaWorker(array $message): string - { - [$topic, $rdMessage] = $message['body']; + private function onKafkaWorker(array $message): string + { + [$topic, $rdMessage] = $message['body']; - call_user_func($message['handler'], new Struct($topic, $rdMessage)); + call_user_func($message['handler'], new Struct($topic, $rdMessage)); - return 'success'; - } + return 'success'; + } } diff --git a/System/Crontab/Zookeeper.php b/System/Crontab/Zookeeper.php index af39bfea..3eabf035 100644 --- a/System/Crontab/Zookeeper.php +++ b/System/Crontab/Zookeeper.php @@ -5,12 +5,12 @@ namespace Snowflake\Crontab; use Exception; +use Snowflake\Cache\Redis; use Snowflake\Process\Process; use Snowflake\Snowflake; -use Swoole\Coroutine\WaitGroup; use Swoole\Coroutine\Channel; +use Swoole\Coroutine\WaitGroup; use Swoole\Timer; -use Snowflake\Cache\Redis; /** * Class Zookeeper @@ -20,40 +20,40 @@ class Zookeeper extends Process { - private Channel $channel; - private WaitGroup $waitGroup; + private Channel $channel; + private WaitGroup $waitGroup; - /** @var Crontab[] $names */ - public array $names = []; + /** @var Crontab[] $names */ + public array $names = []; - public array $scores = []; - public array $timers = []; + public array $scores = []; + public array $timers = []; - /** - * @param \Swoole\Process $process - * @throws Exception - */ - public function onHandler(\Swoole\Process $process): void - { - $crontab = Snowflake::app()->get('crontab'); - $crontab->clearAll(); - if (Snowflake::getPlatform()->isLinux()) { - name($this->pid, 'Crontab zookeeper.'); - } - Timer::tick(1000, function () { - [$range, $redis] = $this->loadCarobTask(); + /** + * @param \Swoole\Process $process + * @throws Exception + */ + public function onHandler(\Swoole\Process $process): void + { + $crontab = Snowflake::app()->get('crontab'); + $crontab->clearAll(); + if (Snowflake::getPlatform()->isLinux()) { + name($this->pid, 'Crontab zookeeper.'); + } + Timer::tick(1000, function () { + [$range, $redis] = $this->loadCarobTask(); - $server = Snowflake::app()->getSwoole(); - $setting = $server->setting['worker_num']; - foreach ($range as $value) { - $this->dispatch($server, $redis, $setting, $value); - } - $redis->release(); - }); - } + $server = Snowflake::app()->getSwoole(); + $setting = $server->setting['worker_num']; + foreach ($range as $value) { + $this->dispatch($server, $redis, $setting, $value); + } + $redis->release(); + }); + } /** @@ -63,47 +63,52 @@ class Zookeeper extends Process * @param $value * @throws Exception */ - private function dispatch($server, Redis|\Redis $redis, int $setting, $value) - { - $server->sendMessage([ - 'action' => 'crontab', 'handler' => swoole_unserialize($redis->get('crontab:' . $value)) - ], random_int(0, $setting - 1)); - } + private function dispatch($server, Redis|\Redis $redis, int $setting, $value) + { + try { + $params['action'] = 'crontab'; + $params['handler'] = $redis->get('crontab:' . $value); + $server->sendMessage($params, random_int(0, $setting - 1)); + } catch (\Throwable $exception) { + logger()->addError($exception); + } + + } - /** - * @return array - * @throws Exception - */ - private function loadCarobTask(): array - { - $redis = Snowflake::app()->getRedis(); + /** + * @return array + * @throws Exception + */ + private function loadCarobTask(): array + { + $redis = Snowflake::app()->getRedis(); - $startTime = time(); + $startTime = time(); - $range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime); - $redis->zRemRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime); + $range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime); + $redis->zRemRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime); - return [$range, $redis]; - } + return [$range, $redis]; + } - /** - * @param string $name - */ - public function clear(string $name) - { - if (!isset($this->names[$name])) { - return; - } - $timers = $this->timers[$name]; + /** + * @param string $name + */ + public function clear(string $name) + { + if (!isset($this->names[$name])) { + return; + } + $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]); - } + $search = array_search($name, $this->scores[$timers]); + if ($search !== false) { + unset($this->scores[$timers][$search]); + } + unset($this->timers[$name], $this->names[$name]); + } }