From 74bcdf257e865b1c56273551e95e40f11dd0484a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mr=C2=B7x?= Date: Mon, 5 Jul 2021 15:42:44 +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/Abstracts/BaseApplication.php | 12 + System/Abstracts/TraitApplication.php | 2 + System/Pool/ClientsPool.php | 314 ++++++++++++++++++++ System/Pool/Connection.php | 403 +++++++++++++------------- System/Pool/Redis.php | 68 ++--- 5 files changed, 559 insertions(+), 240 deletions(-) create mode 100644 System/Pool/ClientsPool.php diff --git a/System/Abstracts/BaseApplication.php b/System/Abstracts/BaseApplication.php index 04a6a986..228a3ee0 100644 --- a/System/Abstracts/BaseApplication.php +++ b/System/Abstracts/BaseApplication.php @@ -38,6 +38,7 @@ use Snowflake\Event; use Snowflake\Exception\InitException; use Snowflake\Exception\NotFindClassException; use Snowflake\Jwt\Jwt; +use Snowflake\Pool\ClientsPool; use Snowflake\Pool\Connection; use Snowflake\Pool\Pool as SPool; use Snowflake\Pool\Redis as SRedis; @@ -434,6 +435,16 @@ abstract class BaseApplication extends Service } + /** + * @return ClientsPool + * @throws Exception + */ + public function getClientsPool(): ClientsPool + { + return $this->get('clientsPool'); + } + + /** * @throws Exception */ @@ -444,6 +455,7 @@ abstract class BaseApplication extends Service 'connections' => ['class' => Connection::class], 'redis_connections' => ['class' => SRedis::class], 'pool' => ['class' => SPool::class], + 'clientsPool' => ['class' => ClientsPool::class], 'config' => ['class' => Config::class], 'logger' => ['class' => Logger::class], 'annotation' => ['class' => SAnnotation::class], diff --git a/System/Abstracts/TraitApplication.php b/System/Abstracts/TraitApplication.php index c1ea821f..c4aa18f2 100644 --- a/System/Abstracts/TraitApplication.php +++ b/System/Abstracts/TraitApplication.php @@ -23,6 +23,7 @@ use Snowflake\Channel; use Snowflake\Error\Logger; use Snowflake\Event; use Snowflake\Jwt\Jwt; +use Snowflake\Pool\ClientsPool; use Snowflake\Pool\Connection; use Snowflake\Pool\Pool as SPool; use Rpc\Producer as RPCProducer; @@ -54,6 +55,7 @@ use Rpc\Producer as RPCProducer; * @property Channel $channel * @property Shutdown $shutdown * @property Emit $emit + * @property ClientsPool $clientsPool */ trait TraitApplication { diff --git a/System/Pool/ClientsPool.php b/System/Pool/ClientsPool.php new file mode 100644 index 00000000..70915b05 --- /dev/null +++ b/System/Pool/ClientsPool.php @@ -0,0 +1,314 @@ +creates); + $this->creates = -1; + } else { + $min = Config::get('databases.pool.min', 1); + if ($this->getChannel($name)->length() > $min) { + $this->flush($min); + } + } + } + + + /** + * @param $retain_number + * @throws Exception + */ + public function flush($retain_number) + { + $channels = $this->getChannels(); + foreach ($channels as $name => $channel) { + $names[] = $name; + $this->pop($channel, $name, $retain_number); + } + static::$_connections = []; + if ($retain_number == 0) { + Timer::clear($this->creates); + $this->creates = -1; + } + } + + + /** + * @param $name + */ + protected function clearCreateLog($name): void + { + if (!isset(static::$hasCreate[$name])) { + return; + } + static::$hasCreate[$name] = 0; + } + + + /** + * @param Channel $channel + * @param $name + * @param $retain_number + * @throws Exception + */ + protected function pop(Channel $channel, $name, $retain_number): void + { + if (Coroutine::getCid() === -1) { + return; + } + while ($channel->length() > $retain_number) { + $connection = $channel->pop(); + if ($connection) { + unset($connection); + } + $this->decrement($name); + } + } + + + /** + * @param $driver + * @param $name + * @param false $isMaster + * @param int $max + */ + public function initConnections($driver, $name, bool $isMaster = false, int $max = 60) + { + $name = $this->name($name, $isMaster); + if (isset(static::$_connections[$name]) && static::$_connections[$name] instanceof Channel) { + return; + } + if (Coroutine::getCid() === -1) { + return; + } + if ($this->creates === -1) { + $this->creates = Timer::tick(1000, [$this, 'Heartbeat_detection'], $name); + } + static::$_connections[$name] = new Channel($max); + $this->max = $max; + } + + + /** + * @param $name + * @return Channel + * @throws ConfigException + */ + private function getChannel($name): Channel + { + if (!isset(static::$_connections[$name])) { + static::$_connections[$name] = new Channel(Config::get('databases.pool.max', 10)); + if ($this->creates === -1) { + $this->creates = Timer::tick(1000, [$this, 'Heartbeat_detection'], $name); + } + } + return static::$_connections[$name]; + } + + + /** + * @param $name + * @return array + * @throws Exception + */ + public function getFromChannel($name): mixed + { + $channel = $this->getChannel($name); + if (!$channel->isEmpty()) { + $connection = $channel->pop(); + if ($this->checkCanUse($name, $connection)) { + return $connection; + } + } + return null; + } + + + /** + * @param $cds + * @param false $isMaster + * @return string + */ + #[Pure] public function name($cds, bool $isMaster = false): string + { + if ($isMaster === true) { + return $cds . '_master'; + } else { + return $cds . '_slave'; + } + } + + + /** + * @param string $name + * @param mixed $client + * @return bool + * 检查连接可靠性 + */ + public function checkCanUse(string $name, mixed $client): bool + { + return true; + } + + + /** + * @param array $config + * @param bool $isMaster + * @return mixed + * @throws Exception + */ + public function get(mixed $config, bool $isMaster): mixed + { + throw new Exception('Undefined system processing function.'); + } + + + /** + * @param string $name + * @return bool + */ + public function hasItem(string $name): bool + { + if (isset(static::$_connections[$name])) { + return !static::$_connections[$name]->isEmpty(); + } + return false; + } + + + /** + * @param string $name + * @return mixed + */ + public function size(string $name): mixed + { + if (Coroutine::getCid() === -1) { + return 0; + } + if (!isset(static::$_connections[$name])) { + return 0; + } + return static::$_connections[$name]->length(); + } + + + /** + * @param string $name + * @param mixed $client + * @throws ConfigException + */ + public function push(string $name, mixed $client) + { + if (Coroutine::getCid() === -1) { + return; + } + $channel = $this->getChannel($name); + if (!$channel->isFull()) { + $channel->push($client); + } + unset($client); + } + + + /** + * @param string $name + * @throws Exception + */ + public function clean(string $name) + { + if (Coroutine::getCid() === -1 || !isset(static::$_connections[$name])) { + return; + } + $channel = static::$_connections[$name]; + $this->pop($channel, $name, 0); + if ($this->creates > -1) { + Timer::clear($this->creates); + } + $channel->close(); + static::$_connections[$name] = null; + } + + + /** + * @return Channel[] + */ + protected function getChannels(): array + { + return static::$_connections; + } + + +} diff --git a/System/Pool/Connection.php b/System/Pool/Connection.php index cd395f94..859e394f 100644 --- a/System/Pool/Connection.php +++ b/System/Pool/Connection.php @@ -6,7 +6,8 @@ namespace Snowflake\Pool; use Exception; use HttpServer\Http\Context; use PDO; -use Snowflake\Abstracts\Pool; +use Snowflake\Abstracts\Component; +use Snowflake\Snowflake; use Swoole\Coroutine; use Swoole\Error; use Swoole\Runtime; @@ -16,227 +17,229 @@ use Throwable; * Class Connection * @package Snowflake\Pool */ -class Connection extends Pool +class Connection extends Component { - - public int $timeout = 1900; - - /** - * @param $timeout - */ - public function setTimeout($timeout) - { - $this->timeout = $timeout; - } + private ?ClientsPool $clientsPool = null; - /** - * @param $value - */ - public function setLength($value) - { - $this->max = $value; - } + /** + * @param $cds + * @return bool + * + * db is in transaction + * @throws Exception + */ + public function inTransaction($cds): bool + { + return Context::getContext('begin_' . $this->getPool()->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 + * @throws Exception + */ + public function beginTransaction($coroutineName) + { + $coroutineName = $this->getPool()->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 + * @throws Exception + */ + public function commit($coroutineName) + { + $coroutineName = $this->getPool()->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 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 + * @throws Exception + */ + public function rollback($coroutineName) + { + $coroutineName = $this->getPool()->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, bool $isMaster = false): mixed - { - $coroutineName = $this->name('mysql', $config['cds'], $isMaster); - if (($pdo = Context::getContext($coroutineName)) instanceof PDO) { - return $pdo; - } - /** @var PDO $connections */ - $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, bool $isMaster = false): mixed + { + $coroutineName = $this->getPool()->name('Mysql:' . $config['cds'], $isMaster); + if (($pdo = Context::getContext($coroutineName)) instanceof PDO) { + return $pdo; + } + /** @var PDO $connections */ + $connections = $this->getPool()->getFromChannel($coroutineName); + if (empty($connections)) { + $connections = $this->createClient($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 - { - if (Coroutine::getCid() === -1) { - Runtime::enableCoroutine(false); - } - $cds = 'mysql:dbname=' . $config['database'] . ';host=' . $config['cds']; - $link = new PDO($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); + } + $cds = 'mysql:dbname=' . $config['database'] . ';host=' . $config['cds']; + $link = new PDO($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->getPool()->name('Mysql:' . $coroutineName, $isMaster); - /** @var PDO $client */ - if (!($client = Context::getContext($coroutineName)) instanceof PDO) { - return; - } - if ($client->inTransaction()) { - $client->commit(); - } - $this->push($coroutineName, $client); - Context::remove($coroutineName); - $this->lastTime = time(); - } + /** @var PDO $client */ + if (!($client = Context::getContext($coroutineName)) instanceof PDO) { + return; + } + if ($client->inTransaction()) { + $client->commit(); + } + $this->getPool()->push($coroutineName, $client); + Context::remove($coroutineName); + } - /** - * @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->getPool()->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->getPool()->decrement($name); + } + return $result; + } + } - /** - * @param $coroutineName - * @param bool $isMaster - * @throws Exception - */ - public function disconnect($coroutineName, bool $isMaster = false) - { - Context::remove($coroutineName); - $coroutineName = $this->name($coroutineName, $isMaster); - $this->clean($coroutineName); - } + /** + * @param $coroutineName + * @param bool $isMaster + * @throws Exception + */ + public function disconnect($coroutineName, bool $isMaster = false) + { + Context::remove($coroutineName); + $coroutineName = $this->getPool()->name('Mysql:' . $coroutineName, $isMaster); + $this->getPool()->clean($coroutineName); + } + + + /** + * @return ClientsPool + * @throws Exception + */ + public function getPool(): ClientsPool + { + if (!$this->clientsPool) { + $this->clientsPool = Snowflake::app()->getClientsPool(); + } + return $this->clientsPool; + } } diff --git a/System/Pool/Redis.php b/System/Pool/Redis.php index 8b1c9797..c0323e07 100644 --- a/System/Pool/Redis.php +++ b/System/Pool/Redis.php @@ -8,9 +8,10 @@ namespace Snowflake\Pool; use Exception; use HttpServer\Http\Context; use Redis as SRedis; -use Snowflake\Abstracts\Pool; +use Snowflake\Abstracts\Component; use Snowflake\Exception\ConfigException; use Snowflake\Exception\RedisConnectException; +use Snowflake\Snowflake; use Swoole\Coroutine; use Swoole\Runtime; @@ -18,19 +19,10 @@ use Swoole\Runtime; * Class RedisClient * @package Snowflake\Snowflake\Pool */ -class Redis extends Pool +class Redis extends Component { - - public int $_create = 0; - - /** - * @param $value - */ - public function setLength($value) - { - $this->max = $value; - } + private ?ClientsPool $clientsPool = null; /** @@ -41,11 +33,18 @@ class Redis extends Pool */ public function get(mixed $config, bool $isMaster = false): mixed { - $coroutineName = $this->name('redis', $config['host'], $isMaster); - if (!Context::hasContext($coroutineName)) { - return Context::setContext($coroutineName, $this->getFromChannel($coroutineName, $config)); + $coroutineName = $this->getPool()->name('Redis:' . $config['host'], $isMaster); + if (Context::hasContext($coroutineName)) { + return Context::getContext($coroutineName); } - return Context::getContext($coroutineName); + if (Coroutine::getCid() === -1) { + return Context::setContext($coroutineName, $this->createClient($coroutineName, $config)); + } + $clients = $this->getPool()->getFromChannel($coroutineName); + if (empty($clients)) { + return Context::setContext($coroutineName, $this->createClient($coroutineName, $config)); + } + return Context::setContext($coroutineName, $clients); } @@ -75,7 +74,7 @@ class Redis extends Pool $redis->setOption(SRedis::OPT_READ_TIMEOUT, $config['read_timeout']); $redis->setOption(SRedis::OPT_PREFIX, $config['prefix']); - $this->increment($name); + $this->getPool()->increment($name); return $redis; } @@ -85,15 +84,16 @@ class Redis extends Pool * @param array $config * @param bool $isMaster * @throws ConfigException + * @throws Exception */ public function release(array $config, bool $isMaster = false) { - $coroutineName = $this->name('redis', $config['host'], $isMaster); + $coroutineName = $this->getPool()->name('Redis:' . $config['host'], $isMaster); if (!Context::hasContext($coroutineName)) { return; } - $this->push($coroutineName, Context::getContext($coroutineName)); + $this->getPool()->push($coroutineName, Context::getContext($coroutineName)); Context::remove($coroutineName); } @@ -104,37 +104,25 @@ class Redis extends Pool */ public function destroy(array $config, bool $isMaster = false) { - $coroutineName = $this->name('redis', $config['host'], $isMaster); + $coroutineName = $this->getPool()->name('Redis:' . $config['host'], $isMaster); if (Context::hasContext($coroutineName)) { - $this->decrement($coroutineName); + $this->getPool()->decrement($coroutineName); } Context::remove($coroutineName); - $this->flush(0); + $this->getPool()->flush(0); } + /** - * @param string $name - * @param mixed $client - * @return bool + * @return ClientsPool * @throws Exception */ - public function checkCanUse(string $name, mixed $client): bool + public function getPool(): ClientsPool { - try { - if (!($client instanceof SRedis)) { - $result = false; - } else { - $result = true; - } - } catch (\Throwable $exception) { - $this->addError($exception, 'redis'); - $result = false; - } finally { - if (!$result) { - $this->decrement($name); - } - return $result; + if (!$this->clientsPool) { + $this->clientsPool = Snowflake::app()->getClientsPool(); } + return $this->clientsPool; }