From 45f5787dfbcaf788f22481a792a4f85f09f6de75 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mr=C2=B7x?= Date: Fri, 7 May 2021 19:38:37 +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 --- System/Pool/Connection.php | 398 +++++++++++++++++++------------------ System/Pool/Redis.php | 11 +- 2 files changed, 209 insertions(+), 200 deletions(-) diff --git a/System/Pool/Connection.php b/System/Pool/Connection.php index ac207990..03caea27 100644 --- a/System/Pool/Connection.php +++ b/System/Pool/Connection.php @@ -9,6 +9,7 @@ use PDO; use Snowflake\Abstracts\Pool; use Swoole\Coroutine; use Swoole\Error; +use Swoole\Runtime; use Throwable; /** @@ -19,226 +20,229 @@ class Connection extends Pool { - public int $timeout = 1900; + public int $timeout = 1900; - /** - * @param $timeout - */ - public function setTimeout($timeout) - { - $this->timeout = $timeout; - } + /** + * @param $timeout + */ + public function setTimeout($timeout) + { + $this->timeout = $timeout; + } - /** - * @param $value - */ - public function setLength($value) - { - $this->max = $value; - } + /** + * @param $value + */ + public function setLength($value) + { + $this->max = $value; + } - /** - * @param $cds - * @return bool - * - * db is in transaction - */ - public function inTransaction($cds): bool - { - return Context::getContext('begin_' . $this->name('mysql', $cds, true)) == 0; - } + /** + * @param $cds + * @return bool + * + * db is in transaction + */ + public function inTransaction($cds): bool + { + return Context::getContext('begin_' . $this->name('mysql', $cds, true)) == 0; + } - /** - * @param $coroutineName - */ - public function beginTransaction($coroutineName) - { - $coroutineName = $this->name('mysql', $coroutineName, true); - if (!Context::hasContext('begin_' . $coroutineName)) { - Context::setContext('begin_' . $coroutineName, 0); - } - Context::increment('begin_' . $coroutineName); - if (!Context::getContext('begin_' . $coroutineName) !== 0) { - return; - } - $connection = Context::getContext($coroutineName); - if ($connection instanceof PDO && !$connection->inTransaction()) { - $connection->beginTransaction(); - } - } + /** + * @param $coroutineName + */ + public function beginTransaction($coroutineName) + { + $coroutineName = $this->name('mysql', $coroutineName, true); + if (!Context::hasContext('begin_' . $coroutineName)) { + Context::setContext('begin_' . $coroutineName, 0); + } + Context::increment('begin_' . $coroutineName); + if (!Context::getContext('begin_' . $coroutineName) !== 0) { + return; + } + $connection = Context::getContext($coroutineName); + if ($connection instanceof PDO && !$connection->inTransaction()) { + $connection->beginTransaction(); + } + } - /** - * @param $coroutineName - */ - public function commit($coroutineName) - { - $coroutineName = $this->name('mysql', $coroutineName, true); - if (!Context::hasContext('begin_' . $coroutineName)) { - return; - } - if (Context::decrement('begin_' . $coroutineName) > 0) { - return; - } - $connection = Context::getContext($coroutineName); - if (!($connection instanceof PDO)) { - return; - } - Context::setContext('begin_' . $coroutineName, 0); - if ($connection->inTransaction()) { - $connection->commit(); - } - } + /** + * @param $coroutineName + */ + public function commit($coroutineName) + { + $coroutineName = $this->name('mysql', $coroutineName, true); + if (!Context::hasContext('begin_' . $coroutineName)) { + return; + } + if (Context::decrement('begin_' . $coroutineName) > 0) { + return; + } + $connection = Context::getContext($coroutineName); + if (!($connection instanceof PDO)) { + return; + } + Context::setContext('begin_' . $coroutineName, 0); + if ($connection->inTransaction()) { + $connection->commit(); + } + } - /** - * @param $name - * @param false $isMaster - * @return array - */ - private function getIndex($name, $isMaster = false): array - { - return [Coroutine::getCid(), $this->name('mysql', $name, $isMaster)]; - } + /** + * @param $name + * @param false $isMaster + * @return array + */ + private function getIndex($name, $isMaster = false): array + { + return [Coroutine::getCid(), $this->name('mysql', $name, $isMaster)]; + } - /** - * @param $coroutineName - */ - public function rollback($coroutineName) - { - $coroutineName = $this->name('mysql', $coroutineName, true); - if (!Context::hasContext('begin_' . $coroutineName)) { - return; - } - if (Context::decrement('begin_' . $coroutineName) > 0) { - return; - } - if (($connection = Context::getContext($coroutineName)) instanceof PDO) { - if ($connection->inTransaction()) { - $connection->rollBack(); - } - } - Context::setContext('begin_' . $coroutineName, 0); - } + /** + * @param $coroutineName + */ + public function rollback($coroutineName) + { + $coroutineName = $this->name('mysql', $coroutineName, true); + if (!Context::hasContext('begin_' . $coroutineName)) { + return; + } + if (Context::decrement('begin_' . $coroutineName) > 0) { + return; + } + if (($connection = Context::getContext($coroutineName)) instanceof PDO) { + if ($connection->inTransaction()) { + $connection->rollBack(); + } + } + Context::setContext('begin_' . $coroutineName, 0); + } - /** - * @param mixed $config - * @param bool $isMaster - * @return mixed - * @throws Exception - */ - public function get(mixed $config, $isMaster = false): mixed - { - $coroutineName = $this->name('mysql', $config['cds'], $isMaster); - if (($pdo = Context::getContext($coroutineName)) instanceof PDO) { - return $pdo; - } - $connections = $this->getFromChannel($coroutineName, $config); - if ($number = Context::getContext('begin_' . $coroutineName, Coroutine::getCid())) { - $number > 0 && $connections->beginTransaction(); - } - return Context::setContext($coroutineName, $connections); - } + /** + * @param mixed $config + * @param bool $isMaster + * @return mixed + * @throws Exception + */ + public function get(mixed $config, $isMaster = false): mixed + { + $coroutineName = $this->name('mysql', $config['cds'], $isMaster); + if (($pdo = Context::getContext($coroutineName)) instanceof PDO) { + return $pdo; + } + $connections = $this->getFromChannel($coroutineName, $config); + if ($number = Context::getContext('begin_' . $coroutineName, Coroutine::getCid())) { + $number > 0 && $connections->beginTransaction(); + } + return Context::setContext($coroutineName, $connections); + } - /** - * @param string $name - * @param mixed $config - * @return PDO - * @throws Exception - */ - public function createClient(string $name, mixed $config): PDO - { - $link = new PDO($config['cds'], $config['username'], $config['password'], [ - PDO::ATTR_EMULATE_PREPARES => false, - PDO::ATTR_CASE => PDO::CASE_NATURAL, - PDO::ATTR_TIMEOUT => $this->timeout, - PDO::MYSQL_ATTR_USE_BUFFERED_QUERY => true, - PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES ' . ($config['charset'] ?? 'utf8mb4') - ]); - $link->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); - $link->setAttribute(PDO::ATTR_STRINGIFY_FETCHES, false); - $link->setAttribute(PDO::ATTR_ORACLE_NULLS, PDO::NULL_EMPTY_STRING); - return $link; - } + /** + * @param string $name + * @param mixed $config + * @return PDO + * @throws Exception + */ + public function createClient(string $name, mixed $config): PDO + { + if (Coroutine::getCid() === -1) { + Runtime::enableCoroutine(false); + } + $link = new PDO($config['cds'], $config['username'], $config['password'], [ + PDO::ATTR_EMULATE_PREPARES => false, + PDO::ATTR_CASE => PDO::CASE_NATURAL, + PDO::ATTR_TIMEOUT => $this->timeout, + PDO::MYSQL_ATTR_USE_BUFFERED_QUERY => true, + PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES ' . ($config['charset'] ?? 'utf8mb4') + ]); + $link->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); + $link->setAttribute(PDO::ATTR_STRINGIFY_FETCHES, false); + $link->setAttribute(PDO::ATTR_ORACLE_NULLS, PDO::NULL_EMPTY_STRING); + return $link; + } - /** - * @param $coroutineName - * @param $isMaster - * @throws Exception - */ - public function release($coroutineName, $isMaster) - { - $coroutineName = $this->name('mysql', $coroutineName, $isMaster); + /** + * @param $coroutineName + * @param $isMaster + * @throws Exception + */ + public function release($coroutineName, $isMaster) + { + $coroutineName = $this->name('mysql', $coroutineName, $isMaster); - /** @var PDO $client */ - if (!($client = Context::getContext($coroutineName)) instanceof PDO) { - return; - } - if ($client->inTransaction()) { - $client->commit(); - } - $this->push($coroutineName, $client); - $this->lastTime = time(); - } + /** @var PDO $client */ + if (!($client = Context::getContext($coroutineName)) instanceof PDO) { + return; + } + if ($client->inTransaction()) { + $client->commit(); + } + $this->push($coroutineName, $client); + $this->lastTime = time(); + } - /** - * @param $coroutineName - * @return bool - */ - private function hasClient($coroutineName): bool - { - return Context::hasContext($coroutineName); - } + /** + * @param $coroutineName + * @return bool + */ + private function hasClient($coroutineName): bool + { + return Context::hasContext($coroutineName); + } - /** - * batch release - * @throws Exception - */ - public function connection_clear() - { - $this->flush(0); - } + /** + * batch release + * @throws Exception + */ + public function connection_clear() + { + $this->flush(0); + } - /** - * @param string $name - * @param mixed $client - * @return bool - * @throws Exception - */ - public function checkCanUse(string $name, mixed $client): bool - { - try { - if (empty($client) || !($client instanceof PDO)) { - $result = false; - } else { - $result = true; - } - } catch (Error | Throwable $exception) { - $result = $this->addError($exception, 'mysql'); - } finally { - if (!$result) { - $this->decrement($name); - } - return $result; - } - } + /** + * @param string $name + * @param mixed $client + * @return bool + * @throws Exception + */ + public function checkCanUse(string $name, mixed $client): bool + { + try { + if (empty($client) || !($client instanceof PDO)) { + $result = false; + } else { + $result = true; + } + } catch (Error | Throwable $exception) { + $result = $this->addError($exception, 'mysql'); + } finally { + if (!$result) { + $this->decrement($name); + } + return $result; + } + } - /** - * @param $coroutineName - * @param bool $isMaster - * @throws Exception - */ - public function disconnect($coroutineName, $isMaster = false) - { - $coroutineName = $this->name($coroutineName, $isMaster); - $this->clean($coroutineName); - } + /** + * @param $coroutineName + * @param bool $isMaster + * @throws Exception + */ + public function disconnect($coroutineName, $isMaster = false) + { + $coroutineName = $this->name($coroutineName, $isMaster); + $this->clean($coroutineName); + } } diff --git a/System/Pool/Redis.php b/System/Pool/Redis.php index 0093b1a2..50f5786f 100644 --- a/System/Pool/Redis.php +++ b/System/Pool/Redis.php @@ -10,6 +10,8 @@ use HttpServer\Http\Context; use Redis as SRedis; use Snowflake\Abstracts\Pool; use Snowflake\Exception\RedisConnectException; +use Swoole\Coroutine; +use Swoole\Runtime; /** * Class RedisClient @@ -36,10 +38,10 @@ class Redis extends Pool */ public function canCreate(string $name): bool { - if (!isset($this->hasCreate[$name])) { - $this->hasCreate[$name] = 0; + if (!isset(static::$hasCreate[$name])) { + static::$hasCreate[$name] = 0; } - return $this->hasCreate[$name] >= $this->max; + return static::$hasCreate[$name] >= $this->max; } @@ -69,6 +71,9 @@ class Redis extends Pool */ public function createClient(string $name, mixed $config): SRedis { + if (Coroutine::getCid() === -1) { + Runtime::enableCoroutine(false); + } $redis = new SRedis(); if (!$redis->pconnect($config['host'], (int)$config['port'], $config['timeout'])) { throw new RedisConnectException(sprintf('The Redis Connect %s::%d Fail.', $config['host'], $config['port']));