From de6e01b058e133291f0e257015238b9a6dce232e Mon Sep 17 00:00:00 2001 From: "as2252258@163.com" Date: Fri, 16 Apr 2021 00:22:26 +0800 Subject: [PATCH] modify --- HttpServer/Events/OnPipeMessage.php | 108 ++++--- System/Crontab/Crontab.php | 435 ++++++++++++++-------------- System/Crontab/Producer.php | 1 - System/Crontab/Zookeeper.php | 4 +- 4 files changed, 273 insertions(+), 275 deletions(-) diff --git a/HttpServer/Events/OnPipeMessage.php b/HttpServer/Events/OnPipeMessage.php index cd9fa3b6..68d5da9d 100644 --- a/HttpServer/Events/OnPipeMessage.php +++ b/HttpServer/Events/OnPipeMessage.php @@ -18,70 +18,68 @@ use Swoole\Server; class OnPipeMessage extends Callback { - /** - * @param Server $server - * @param int $src_worker_id - * @param $swollen_universalize - * @throws Exception - */ - public function onHandler(Server $server, int $src_worker_id, $swollen_universalize) - { - go(function () use ($server, $src_worker_id, $swollen_universalize) { - match ($swollen_universalize['action'] ?? null) { - 'kafka' => $this->onKafkaWorker($swollen_universalize), - 'crontab' => $this->onCrontabWorker($swollen_universalize), - default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize) - }; - fire(Event::SYSTEM_RESOURCE_RELEASES); - }); - } + /** + * @param Server $server + * @param int $src_worker_id + * @param $swollen_universalize + * @throws Exception + */ + public function onHandler(Server $server, int $src_worker_id, $swollen_universalize) + { + go(function () use ($server, $src_worker_id, $swollen_universalize) { + match ($swollen_universalize['action'] ?? null) { + 'kafka' => $this->onKafkaWorker($swollen_universalize), + 'crontab' => $this->onCrontabWorker($swollen_universalize), + default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize) + }; + fire(Event::SYSTEM_RESOURCE_RELEASES); + }); + } - /** - * @param array $message - * @return string - * @throws Exception - */ - private function onCrontabWorker(array $message): string - { - if (!isset($message['handler'])) { - throw new Exception('unknown handler'); - } - - /** @var Crontab $crontab */ - $crontab = swoole_unserialize($message['handler']); - $crontab->increment()->execute(); - return 'success'; - } + /** + * @param array $message + * @return string + * @throws Exception + */ + private function onCrontabWorker(array $message): string + { + if (empty($handler = $message['handler'] ?? null)) { + throw new Exception('unknown handler'); + } + /** @var Crontab $handler */ + $handler->increment()->execute(); + return 'success'; + } - /** - * @param $server - * @param $src_worker_id - * @param $message - * @return string - * @throws Exception - */ - private function onMessageWorker($server, $src_worker_id, $message): string - { - fire(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]); + /** + * @param $server + * @param $src_worker_id + * @param $message + * @return string + * @throws Exception + */ + 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']; + /** + * @param array $message + * @return string + */ + 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/Crontab.php b/System/Crontab/Crontab.php index e4cfb719..514eda2c 100644 --- a/System/Crontab/Crontab.php +++ b/System/Crontab/Crontab.php @@ -18,277 +18,278 @@ use Swoole\Timer; abstract class Crontab extends BaseObject { - const WAIT_END = 'crontab:wait:execute'; + const WAIT_END = 'crontab:wait:execute'; - 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 $timerId = -1; + private int $timerId = -1; - private int $max_execute_number = -1; + private int $max_execute_number = -1; - private int $execute_number = 0; + private int $execute_number = 0; - /** - * @return $this - */ - public function increment(): static - { - $this->execute_number += 1; - return $this; - } + /** + * @return $this + */ + public function increment(): static + { + $this->execute_number += 1; + return $this; + } - /** - * @return string - */ - #[Pure] public function getName(): string - { - return md5($this->name); - } + /** + * @return string + */ + #[Pure] public function getName(): string + { + return md5($this->name); + } - /** - * @return mixed - */ - public function getParams(): mixed - { - return $this->params; - } + /** + * @return mixed + */ + public function getParams(): mixed + { + return $this->params; + } - /** - * @return int - */ - public function getTickTime(): int - { - return $this->tickTime; - } + /** + * @return int + */ + public function getTickTime(): int + { + return $this->tickTime; + } - /** - * @return bool - */ - public function isLoop(): bool - { - return $this->isLoop; - } + /** + * @return bool + */ + public function isLoop(): bool + { + return $this->isLoop; + } - /** - * @return int - */ - public function getMaxExecuteNumber(): int - { - return $this->max_execute_number; - } + /** + * @return int + */ + public function getMaxExecuteNumber(): int + { + return $this->max_execute_number; + } - /** - * @return int - */ - public function getExecuteNumber(): int - { - return $this->execute_number; - } + /** + * @return int + */ + public function getExecuteNumber(): int + { + return $this->execute_number; + } - /** - * @param string $name - */ - public function setName(string $name): void - { - $this->name = $name; - } + /** + * @param string $name + */ + public function setName(string $name): void + { + $this->name = $name; + } - public function setParams(): void - { - $this->params = func_get_args(); - } + public function setParams(): void + { + $this->params = func_get_args(); + } - /** - * @param int $tickTime - */ - public function setTickTime(int $tickTime): void - { - $this->tickTime = $tickTime; - } + /** + * @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 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 $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; - } + /** + * @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; - } + /** + * @return int + */ + public function getTimerId(): int + { + return $this->timerId; + } - /** - * @param int $timerId - */ - public function setTimerId(int $timerId): void - { - $this->timerId = $timerId; - } + /** + * @param int $timerId + */ + public function setTimerId(int $timerId): void + { + $this->timerId = $timerId; + } - /** - * - * @throws Exception - */ - public function clearTimer() - { - $this->warning('crontab timer clear.'); - if (Timer::exists($this->timerId)) { - Timer::clear($this->timerId); - } - } + /** + * + * @throws Exception + */ + public function clearTimer() + { + $this->warning('crontab timer clear.'); + if (Timer::exists($this->timerId)) { + Timer::clear($this->timerId); + } + } - /** - * @throws Exception - */ - private function recover() - { - $redis = Snowflake::app()->getRedis(); - if (!$redis->exists('stop:crontab:' . $this->getName())) { - $redis->del('stop:crontab:' . $this->getName()); - return; - } - - $redis->set('crontab:' . ($name = $this->getName()), swoole_serialize($this)); - - $tickTime = time() + $this->getTickTime(); - - $redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name); - } + /** + * @throws Exception + */ + private function recover() + { + $redis = Snowflake::app()->getRedis(); + if (!$redis->exists('stop:crontab:' . $this->getName())) { + $redis->del('crontab:' . $this->getName()); + $redis->del('stop:crontab:' . $this->getName()); + } else { + $redis->set('crontab:' . ($name = $this->getName()), swoole_serialize($this)); + $tickTime = time() + $this->getTickTime(); + $redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name); + } + } - abstract public function process(): mixed; + abstract public function process(): mixed; - abstract public function max_execute(): mixed; + abstract public function max_execute(): mixed; - abstract public function isStop(): bool; + abstract public function isStop(): bool; - /** - * @throws Exception - */ - public function execute(): void - { - \Swoole\Coroutine\go(function ($application) { - $application->isRecover($application); - }, $this); - $this->run(); - } + /** + * @throws Exception + */ + public function execute(): void + { + \Swoole\Coroutine\go(function ($application) { + $application->isRecover($application); + }, $this); + $this->run(); + } - /** - * @param $application - * @throws Exception - */ - public function isRecover($application) - { - try { - if ($application->isExit()) { - return; - } - if ($application->isMaxExecute()) { - call_user_func([$application, 'max_execute'], ...$application->params); - } else { - $application->recover(); - } - } catch (\Throwable $throwable) { - logger()->addError($throwable, 'throwable'); - } - } + /** + * @param $application + * @return bool|int + * @throws \Exception + */ + public function isRecover($application): bool|int + { + try { + $redis = Snowflake::app()->getRedis(); + if ($application->isExit()) { + return $redis->del('crontab:' . $this->getName()); + } + if ($application->isMaxExecute()) { + call_user_func([$application, 'max_execute'], ...$application->params); + return $redis->del('crontab:' . $this->getName()); + } else { + return $application->recover(); + } + } catch (\Throwable $throwable) { + return logger()->addError($throwable, 'throwable'); + } + } - /** - * @throws Exception - */ - private function run() - { - try { - $redis = Snowflake::app()->getRedis(); + /** + * @throws Exception + */ + private function run() + { + try { + $redis = Snowflake::app()->getRedis(); - $name_md5 = $this->getName(); + $name_md5 = $this->getName(); - $redis->hSet(self::WAIT_END, $name_md5, serialize($this)); + $redis->hSet(self::WAIT_END, $name_md5, serialize($this)); - $params = call_user_func([$this, 'process'], ...$this->params); - $redis->hDel(self::WAIT_END, $name_md5); - if ($params === null) { - return; - } - $name = date('Y-m-d.log'); - write(storage($name, '/log/crontab'), Json::encode([ - 'name' => $this->name, - 'response' => serialize($params) - ])); - } catch (\Throwable $throwable) { - logger()->addError($throwable, 'throwable'); - } - } + $params = call_user_func([$this, 'process'], ...$this->params); + $redis->hDel(self::WAIT_END, $name_md5); + if ($params === null) { + return; + } + $name = date('Y-m-d.log'); + write(storage($name, '/log/crontab'), Json::encode([ + 'name' => $this->name, + 'response' => serialize($params) + ])); + } catch (\Throwable $throwable) { + logger()->addError($throwable, 'throwable'); + } + } - /** - * @return bool - */ - private function isExit(): bool - { - if ($this->isStop()) { - return true; - } - if (!$this->isLoop) { - return true; - } - return false; - } + /** + * @return bool + */ + private function isExit(): bool + { + if ($this->isStop()) { + return true; + } + if (!$this->isLoop) { + return true; + } + return false; + } - /** - * @return bool - */ - private function isMaxExecute(): bool - { - if ($this->max_execute_number !== -1) { - return $this->execute_number >= $this->max_execute_number; - } - return false; - } + /** + * @return bool + */ + private function isMaxExecute(): bool + { + if ($this->max_execute_number !== -1) { + return $this->execute_number >= $this->max_execute_number; + } + return false; + } } diff --git a/System/Crontab/Producer.php b/System/Crontab/Producer.php index 20aa1eec..f612c238 100644 --- a/System/Crontab/Producer.php +++ b/System/Crontab/Producer.php @@ -35,7 +35,6 @@ class Producer extends Component } $tickTime = time() + $crontab->getTickTime(); - $redis->del('stop:crontab:' . $name, 120); $result = $redis->zAdd(self::CRONTAB_KEY, $tickTime, $name); diff --git a/System/Crontab/Zookeeper.php b/System/Crontab/Zookeeper.php index d9cc2ffb..85ce7f03 100644 --- a/System/Crontab/Zookeeper.php +++ b/System/Crontab/Zookeeper.php @@ -67,8 +67,8 @@ class Zookeeper extends Process { try { $params['action'] = 'crontab'; - $params['handler'] = $redis->get('crontab:' . $value); - $redis->del('crontab:' . $value); + $params['handler'] = swoole_unserialize($redis->get('crontab:' . $value)); + $result = $server->sendMessage($params, $workerId = random_int(0, $setting - 1)); var_dump('send crontab to ' . $workerId . ' ' . intval($result));