From 978cc92dad9e86a5279fe15f1b0a35d2e7693202 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mr=C2=B7x?= Date: Thu, 22 Apr 2021 11:56:33 +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 --- Kafka/Producer.php | 39 ++-- System/Crontab/Crontab.php | 460 +++++++++++++++++++------------------ 2 files changed, 254 insertions(+), 245 deletions(-) diff --git a/Kafka/Producer.php b/Kafka/Producer.php index a3eebceb..0378dd6e 100644 --- a/Kafka/Producer.php +++ b/Kafka/Producer.php @@ -11,6 +11,7 @@ use ReflectionException; use Snowflake\Abstracts\Component; use Snowflake\Abstracts\Config; use Snowflake\Event; +use Snowflake\Exception\ConfigException; use Snowflake\Exception\NotFindClassException; use Snowflake\Snowflake; @@ -144,13 +145,12 @@ class Producer extends Component } - /** - * @param $topic - * @param $groupId - * @param $brokers - * @return ProducerTopic - * @throws \Snowflake\Exception\ConfigException - */ + /** + * @param $topic + * @param $groupId + * @throws ConfigException + * @throws Exception + */ private function beforePushMessage($topic, $groupId): void { $consumers = Config::get('kafka.producers.' . $topic); @@ -171,13 +171,12 @@ class Producer extends Component } - /** - * @param TopicConf $topicConf - * @param string $topic - * @param array $message - * @param string $key - * @throws Exception - */ + /** + * @param string $topic + * @param array $message + * @param string $key + * @throws Exception + */ private function sendMessage(string $topic, array $message, string $key = '') { $producer = $this->getProducer(); @@ -205,12 +204,12 @@ class Producer extends Component } - - /** - * @param \RdKafka\Producer $producer - * @param ProducerTopic $producerTopic - * @throws Exception - */ + /** + * @param string $topic + * @param \RdKafka\Producer $producer + * @param ProducerTopic $producerTopic + * @throws Exception + */ private function recover(string $topic, \RdKafka\Producer $producer, ProducerTopic $producerTopic) { $channel = Snowflake::app()->getChannel(); diff --git a/System/Crontab/Crontab.php b/System/Crontab/Crontab.php index 31f95b8b..449b37bc 100644 --- a/System/Crontab/Crontab.php +++ b/System/Crontab/Crontab.php @@ -18,282 +18,292 @@ 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(): bool - { - $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); - } - return true; - } + /** + * @throws Exception + */ + private function recover(): bool + { + $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()), static::getSerialize($this)); + $tickTime = time() + $this->getTickTime(); + $redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name); + } + return true; + } - 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 - { - defer(function () { - $this->isRecover(); - }); - $this->run(); - } + /** + * @throws Exception + */ + public function execute(): void + { + defer(function () { + $this->isRecover(); + }); + $this->run(); + } - /** - * @return bool|int - * @throws Exception - */ - public function isRecover(): bool|int - { - try { - $redis = Snowflake::app()->getRedis(); - if ($redis->exists('stop:crontab:' . $this->getName())) { - $redis->del('stop:crontab:' . $this->getName()); - return true; - } - if ($this->isExit()) { - return $redis->del('crontab:' . $this->getName()); - } - if ($this->isMaxExecute()) { - call_user_func([$this, 'max_execute'], ...$this->getParams()); - return $redis->del('crontab:' . $this->getName()); - } else { - return $this->recover(); - } - } catch (\Throwable $throwable) { - return logger()->addError($throwable, 'throwable'); - } - } + /** + * @return bool|int + * @throws Exception + */ + public function isRecover(): bool|int + { + try { + $redis = Snowflake::app()->getRedis(); + if ($redis->exists('stop:crontab:' . $this->getName())) { + $redis->del('stop:crontab:' . $this->getName()); + return true; + } + if ($this->isExit()) { + return $redis->del('crontab:' . $this->getName()); + } + if ($this->isMaxExecute()) { + call_user_func([$this, 'max_execute'], ...$this->getParams()); + return $redis->del('crontab:' . $this->getName()); + } else { + return $this->recover(); + } + } catch (\Throwable $throwable) { + return logger()->addError($throwable, 'throwable'); + } + } - /** - * @throws Exception - */ - private function run() - { - try { - $redis = Snowflake::app()->getRedis(); - - $name_md5 = $this->getName(); - - $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'); - } - } + /** + * @param $class + * @return string + */ + public static function getSerialize($class): string + { + return serialize($class); + } - /** - * @return bool - */ - private function isExit(): bool - { - if ($this->isStop()) { - return true; - } - if (!$this->isLoop) { - return true; - } - return false; - } + /** + * @throws Exception + */ + private function run() + { + try { + $redis = Snowflake::app()->getRedis(); + + $name_md5 = $this->getName(); + + $redis->hSet(self::WAIT_END, $name_md5, static::getSerialize($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'); + } + } - /** - * @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 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; + } }