From 6cb8172b56c7488589a5ba658f40e684afa397b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mr=C2=B7x?= Date: Tue, 10 Nov 2020 17:52:30 +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 --- Queue/Abstracts/AbstractsQueue.php | 17 --- Queue/Abstracts/Queue.php | 75 ------------- Queue/Abstracts/Relyon.php | 33 ------ Queue/Complete.php | 46 -------- Queue/Consumer.php | 25 ----- Queue/Queue.php | 164 ----------------------------- Queue/QueueProviders.php | 32 ------ Queue/Running.php | 48 --------- Queue/TestQueue.php | 49 --------- Queue/Waiting.php | 46 -------- System/Application.php | 4 - System/Cache/Redis.php | 2 - composer.json | 3 +- 13 files changed, 1 insertion(+), 543 deletions(-) delete mode 100644 Queue/Abstracts/AbstractsQueue.php delete mode 100644 Queue/Abstracts/Queue.php delete mode 100644 Queue/Abstracts/Relyon.php delete mode 100644 Queue/Complete.php delete mode 100644 Queue/Consumer.php delete mode 100644 Queue/Queue.php delete mode 100644 Queue/QueueProviders.php delete mode 100644 Queue/Running.php delete mode 100644 Queue/TestQueue.php delete mode 100644 Queue/Waiting.php diff --git a/Queue/Abstracts/AbstractsQueue.php b/Queue/Abstracts/AbstractsQueue.php deleted file mode 100644 index 5005bcbd..00000000 --- a/Queue/Abstracts/AbstractsQueue.php +++ /dev/null @@ -1,17 +0,0 @@ -getRedis(); - try { - $serialize = serialize($consumer); - if (!$redis->lock($hash = md5($serialize))) { - return false; - } - $isExists = $redis->zRevRank($key, $serialize); - if ($isExists !== null) { - $redis->zAdd($key, $score, $serialize); - } - return $redis->unlink($hash); - } finally { - $redis->release(); - } - } - - - /** - * @param $key - * @param string $consumer - * @return false|int - * @throws ComponentException - * @throws Exception - */ - protected function pop($key, string $consumer) - { - $redis = Snowflake::app()->getRedis(); - try { - if (!$redis->lock($hash = md5($consumer))) { - return false; - } - $isExists = $redis->zRevRank($key, $consumer); - if ($isExists === null) { - return $redis->unlink($hash); - } - $redis->zRem($key, $consumer); - return $redis->unlink($hash); - } finally { - $redis->release(); - } - } - - -} diff --git a/Queue/Abstracts/Relyon.php b/Queue/Abstracts/Relyon.php deleted file mode 100644 index 737c4e55..00000000 --- a/Queue/Abstracts/Relyon.php +++ /dev/null @@ -1,33 +0,0 @@ -push(self::QUEUE_COMPLETE, $consumer, $score); - } - - - /** - * @param $consumer - * @return false|int - * @throws ComponentException - */ - public function del(string $consumer) - { - return $this->pop(self::QUEUE_COMPLETE, $consumer); - } - - -} diff --git a/Queue/Consumer.php b/Queue/Consumer.php deleted file mode 100644 index 34f29aa9..00000000 --- a/Queue/Consumer.php +++ /dev/null @@ -1,25 +0,0 @@ -waiting = Snowflake::createObject(Waiting::class); - $this->running = Snowflake::createObject(Running::class); - $this->complete = Snowflake::createObject(Complete::class); - } - - - /** - * @param Process $process - */ - public function onHandler(Process $process) - { - Timer::tick(50, function () { - $redis = Snowflake::app()->getRedis(); - try { - $params = $redis->zRevRange(Waiting::QUEUE_WAITING, 0, 20); - if (empty($params)) { - return; - } - Coroutine::create(function () use ($params) { - $this->scheduler($params); - }); - } catch (\Throwable $exception) { - $this->application->error($exception->getMessage()); - } finally { - $redis->release(); - } - }); - } - - - /** - * @param Consumer $consumer - * @param int $score - * @throws ComponentException - */ - public function delivery(Consumer $consumer, $score = 0) - { - try { - $consumer->onWaiting(); - } catch (\Throwable $exception) { - $this->application->error($exception->getMessage()); - } finally { - $this->waiting->add($consumer, $score); - } - } - - - /** - * @param array $data - * @throws Exception - */ - private function scheduler(array $data) - { - $redis = Snowflake::app()->getRedis(); - foreach ($data as $datum) { - $this->runner($datum); - } - $redis->release(); - if ($this->shutdown === true) { - Snowflake::shutdown($this); - } - } - - - /** - * @param $class - * @return mixed|void - * @throws ComponentException - */ - private function runner(string $class) - { - try { - $rely_on = unserialize($class); - $this->waiting->del($class); - if (!($rely_on instanceof Consumer)) { - return; - } - $this->running->add($rely_on); - $rely_on->onRunning(); - } catch (\Throwable $exception) { - logger()->write($exception->getMessage(), 'queue'); - } finally { - $this->running->del($class); - if (isset($rely_on) && $rely_on instanceof Consumer) { - $rely_on->onComplete(); - $this->complete->add($rely_on); - } - } - } - - - /** - * @return Waiting - */ - public function getWaiting(): Waiting - { - return $this->waiting; - } - - /** - * @return Complete - */ - public function getComplete(): Complete - { - return $this->complete; - } - - /** - * @return Running - */ - public function getRunning(): Running - { - return $this->running; - } - -} diff --git a/Queue/QueueProviders.php b/Queue/QueueProviders.php deleted file mode 100644 index 047d7372..00000000 --- a/Queue/QueueProviders.php +++ /dev/null @@ -1,32 +0,0 @@ -get('server'); - $server->addProcess('queue', Queue::class); - } - -} diff --git a/Queue/Running.php b/Queue/Running.php deleted file mode 100644 index c73a20ce..00000000 --- a/Queue/Running.php +++ /dev/null @@ -1,48 +0,0 @@ -push(self::QUEUE_RUNNING, $consumer, $score); - } - - - /** - * @param $consumer - * @return false|int - * @throws ComponentException - */ - public function del(string $consumer) - { - return $this->pop(self::QUEUE_RUNNING, $consumer); - } - - -} diff --git a/Queue/TestQueue.php b/Queue/TestQueue.php deleted file mode 100644 index a5b69f50..00000000 --- a/Queue/TestQueue.php +++ /dev/null @@ -1,49 +0,0 @@ -params = $params; - $this->onWaiting(); - } - - /** - * - */ - public function onWaiting() - { - - - - } - - public function onRunning() - { - // TODO: Implement onRunning() method. - } - - public function onComplete() - { - // TODO: Implement onComplete() method. - } -} diff --git a/Queue/Waiting.php b/Queue/Waiting.php deleted file mode 100644 index e7761775..00000000 --- a/Queue/Waiting.php +++ /dev/null @@ -1,46 +0,0 @@ -push(self::QUEUE_WAITING, $consumer, $score); - } - - - /** - * @param $consumer - * @return false|int - * @throws ComponentException - */ - public function del(string $consumer) - { - return $this->pop(self::QUEUE_WAITING, $consumer); - } - -} diff --git a/System/Application.php b/System/Application.php index 40fb3274..476fe3c7 100644 --- a/System/Application.php +++ b/System/Application.php @@ -15,7 +15,6 @@ use Console\ConsoleProviders; use Database\DatabasesProviders; use Exception; use HttpServer\ServerProviders; -use Queue\QueueProviders; use Snowflake\Abstracts\BaseApplication; use Snowflake\Abstracts\Config; use Snowflake\Abstracts\Input; @@ -50,9 +49,6 @@ class Application extends BaseApplication $this->import(ConsoleProviders::class); $this->import(DatabasesProviders::class); $this->import(ServerProviders::class); - if (Config::get('queue.enable', false, false)) { - $this->import(QueueProviders::class); - } } diff --git a/System/Cache/Redis.php b/System/Cache/Redis.php index 9582373f..9bf81f60 100644 --- a/System/Cache/Redis.php +++ b/System/Cache/Redis.php @@ -14,9 +14,7 @@ use Snowflake\Abstracts\Component; use Snowflake\Abstracts\Config; use Snowflake\Event; use Snowflake\Exception\ConfigException; -use Snowflake\Exception\RedisConnectException; use Snowflake\Snowflake; -use Swoole\Coroutine; /** * Class Redis diff --git a/composer.json b/composer.json index 39d41e5d..39c0f376 100644 --- a/composer.json +++ b/composer.json @@ -9,7 +9,7 @@ ], "license": "MIT", "require": { - "php": ">=7.4", + "php": ">=8.0", "swoole/ide-helper": "@dev", "ext-json": "*", "phpmailer/phpmailer": "^6.1", @@ -39,7 +39,6 @@ "validator\\": "Validator/", "Console\\": "Console/", "Database\\": "Database/", - "Queue\\": "Queue/", "Gii\\": "Gii/", "Kafka\\": "Kafka/" },