From f08fd0a7b8f857340d17ff8d18dff832b7aefe80 Mon Sep 17 00:00:00 2001 From: "as2252258@163.com" Date: Mon, 12 Apr 2021 02:55:12 +0800 Subject: [PATCH] modify --- HttpServer/Events/OnPipeMessage.php | 18 ++- Kafka/TaskContainer.php | 1 - System/Crontab/Consumer.php | 129 ------------------ System/Crontab/Crontab.php | 48 ++++++- System/Crontab/CrontabProviders.php | 3 +- .../{ZookeeperProcess.php => Zookeeper.php} | 57 +++++--- 6 files changed, 99 insertions(+), 157 deletions(-) delete mode 100644 System/Crontab/Consumer.php rename System/Crontab/{ZookeeperProcess.php => Zookeeper.php} (54%) diff --git a/HttpServer/Events/OnPipeMessage.php b/HttpServer/Events/OnPipeMessage.php index 4fcde242..2973dd43 100644 --- a/HttpServer/Events/OnPipeMessage.php +++ b/HttpServer/Events/OnPipeMessage.php @@ -4,15 +4,11 @@ declare(strict_types=1); namespace HttpServer\Events; -use Annotation\Loader; use Exception; use HttpServer\Abstracts\Callback; -use Kafka\ConsumerInterface; use Kafka\Struct; -use Kafka\TaskContainer; use Snowflake\Event; use Snowflake\Exception\ComponentException; -use Snowflake\Snowflake; use Swoole\Server; /** @@ -35,6 +31,7 @@ class OnPipeMessage extends Callback $swoole_unserialize = swoole_unserialize($message); match ($swoole_unserialize['action'] ?? null) { 'kafka' => $this->onKafkaWorker($swoole_unserialize), + 'crontab' => $this->onCrontabWorker($swoole_unserialize), default => $this->onMessageWorker($server, $src_worker_id, $message) }; } catch (\Throwable $exception) { @@ -45,6 +42,19 @@ class OnPipeMessage extends Callback } + /** + * @param array $message + * @return string + */ + private function onCrontabWorker(array $message) + { + /** @var \Snowflake\Crontab\Crontab $crontab */ + $crontab = $message['handler'] ?? null; + $crontab->increment()->execute(); + return 'success'; + } + + /** * @param $server * @param $src_worker_id diff --git a/Kafka/TaskContainer.php b/Kafka/TaskContainer.php index e3c90caa..fb1b78ca 100644 --- a/Kafka/TaskContainer.php +++ b/Kafka/TaskContainer.php @@ -48,7 +48,6 @@ class TaskContainer extends BaseObject public function process($topic, Struct $struct) { $handler = $this->_topics[$topic] ?? null; - var_dump($this->_topics, $topic, $struct); if (empty($handler)) { return; } diff --git a/System/Crontab/Consumer.php b/System/Crontab/Consumer.php deleted file mode 100644 index 273a65fb..00000000 --- a/System/Crontab/Consumer.php +++ /dev/null @@ -1,129 +0,0 @@ -isLinux()) { - name($this->pid, 'Crontab consumer'); - } - - $this->channel = new Coroutine\Channel(2000); - go(function () { - $this->popChannel(); - }); - - $this->recovery(); - $this->tick($process); - } - - - private function recovery() - { - $redis = redis()->hGetAll('crontab:wait:execute'); - foreach ($redis as $redi) { - $this->channel->push($redi); - } - fire(Event::SYSTEM_RESOURCE_RELEASES); - } - - - /** - * @throws Exception - */ - public function popChannel() - { - /** @var Crontab $crontab */ - $crontab = $this->channel->pop(-1); - go(function () use ($crontab) { - try { - $crontab->increment()->execute(); - if ($crontab->getExecuteNumber() < $crontab->getMaxExecuteNumber()) { - Consumer::addTask($crontab); - } else if ($crontab->isLoop()) { - Consumer::addTask($crontab); - } - } catch (\Throwable $throwable) { - logger()->addError($throwable, 'throwable'); - } finally { - fire(Event::SYSTEM_RESOURCE_RELEASES); - } - }); - $this->popChannel(); - } - - - /** - * @param \Swoole\Process $process - * @throws ReflectionException - * @throws ComponentException - * @throws ConfigException - * @throws NotFindClassException - * @throws Exception - */ - public function tick(\Swoole\Process $process) - { - $value = $process->read(40); - - $redis = Snowflake::app()->getRedis(); - - /** @var Crontab $crontab */ - $crontab = swoole_unserialize($redis->get($value)); - $redis->hSet('crontab:wait:execute', $crontab->getName(), $redis->get($value)); - $redis->del($value); - if (is_object($crontab)) { - $this->channel->push($crontab); - } - - $redis->release(); - - $this->tick($process); - } - - - /** - * @param Crontab $crontab - * @throws Exception - */ - private static function addTask(Crontab $crontab) - { - $redis = Snowflake::app()->getRedis(); - - $name = md5($crontab->getName()); - - $redis->set('crontab:' . $name, swoole_serialize($crontab)); - - $tickTime = time() + $crontab->getTickTime(); - - $redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $crontab->getName()); - } - - -} diff --git a/System/Crontab/Crontab.php b/System/Crontab/Crontab.php index 6c8e2f00..f47f8414 100644 --- a/System/Crontab/Crontab.php +++ b/System/Crontab/Crontab.php @@ -8,6 +8,7 @@ use Closure; use Exception; use Snowflake\Abstracts\BaseObject; use Snowflake\Event; +use Snowflake\Snowflake; use Swoole\Timer; /** @@ -17,6 +18,7 @@ use Swoole\Timer; class Crontab extends BaseObject { + const WAIT_END = 'crontab:wait:execute'; private array|Closure $handler; @@ -194,16 +196,56 @@ class Crontab extends BaseObject } + /** + * @throws \Exception + */ + private function recover() + { + $redis = Snowflake::app()->getRedis(); + + $redis->set('crontab:' . md5($this->getName()), swoole_serialize($this)); + + $tickTime = time() + $this->getTickTime(); + + $redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $this->getName()); + } + + /** * @throws Exception */ public function execute(): void { - $params = call_user_func($this->handler, $this->params, $this->name); - \redis()->hDel('crontab:wait:execute', $this->getName()); - if ($params !== null) { + try { + $redis = Snowflake::app()->getRedis(); + + $name_md5 = md5($this->getName()); + + $redis->hSet(self::WAIT_END, $name_md5, serialize($this)); + $params = call_user_func($this->handler, $this->params, $this->name); + $redis->hDel(self::WAIT_END, $name_md5); + if ($params === null) { + return; + } $name = date('Y_m_d_H_i_s.' . $this->name . '.log'); write(storage($name, '/log/crontab'), serialize($params)); + } catch (\Throwable $throwable) { + logger()->addError($throwable, 'throwable'); + } finally { + $this->after(); + } + } + + + /** + * @throws \Exception + */ + private function after(): void + { + if ($this->getExecuteNumber() < $this->getMaxExecuteNumber()) { + $this->recover(); + } else if ($this->isLoop()) { + $this->recover(); } } diff --git a/System/Crontab/CrontabProviders.php b/System/Crontab/CrontabProviders.php index d6ad4eb3..616d590e 100644 --- a/System/Crontab/CrontabProviders.php +++ b/System/Crontab/CrontabProviders.php @@ -37,8 +37,7 @@ class CrontabProviders extends Providers if (Config::get('crontab.enable') !== true) { return; } - $server->addProcess('CrontabZookeeper', ZookeeperProcess::class); - $server->addProcess('Consumer', Consumer::class); + $server->addProcess('CrontabZookeeper', Zookeeper::class); } } diff --git a/System/Crontab/ZookeeperProcess.php b/System/Crontab/Zookeeper.php similarity index 54% rename from System/Crontab/ZookeeperProcess.php rename to System/Crontab/Zookeeper.php index f2a19c8b..55aa9480 100644 --- a/System/Crontab/ZookeeperProcess.php +++ b/System/Crontab/Zookeeper.php @@ -12,10 +12,10 @@ use Swoole\Coroutine\Channel; use Swoole\Timer; /** - * Class ZookeeperProcess + * Class Zookeeper * @package Snowflake\Process */ -class ZookeeperProcess extends Process +class Zookeeper extends Process { @@ -39,32 +39,53 @@ class ZookeeperProcess extends Process { $crontab = Snowflake::app()->get('crontab'); $crontab->clearAll(); - if (Snowflake::getPlatform()->isLinux()) { name($this->pid, 'Crontab zookeeper.'); } Timer::tick(1000, function () { - $startTime = time(); - - $redis = Snowflake::app()->getRedis(); - - $redis->multi(); - $range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime); - $redis->zRemRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime); - $redis->exec(); - - $redis->release(); - - /** @var Consumer $consumer */ - $consumer = Snowflake::app()->get(Consumer::class); - + [$range, $redis] = $this->loadCrotabTask(); + $server = Snowflake::app()->getSwoole(); + $setting = $server->setting['worker_num']; foreach ($range as $value) { - $consumer->write('crontab:' . md5($value)); + $this->dispatch($server, $redis, $setting, $value); } + $redis->release(); }); } + /** + * @param $server + * @param $redis + * @throws \Exception + */ + private function dispatch($server, $redis, $setting, $value) + { + $server->sendMessage(swoole_serialize([ + 'action' => 'crontab', 'handler' => swoole_unserialize($redis->get('crontab:' . md5($value))) + ]), random_int(0, $setting - 1)); + $redis->del('crontab:' . md5($value)); + } + + + /** + * @return array + * @throws \Exception + */ + private function loadCrotabTask() + { + $redis = Snowflake::app()->getRedis(); + + $startTime = time(); + + $redis->multi(); + $range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime); + $redis->zRemRangeByScore(Producer::CRONTAB_KEY, '0', (string)$startTime); + $redis->exec(); + + return [$range, $redis]; + } + /** * @param string $name