From 8cd5562b20f2b90ad043f3602987d08a1a9d4b14 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mr=C2=B7x?= Date: Fri, 23 Apr 2021 18:37:18 +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/Pool.php | 69 ++++++++-- System/Pool/Connection.php | 54 +++----- System/Pool/Redis.php | 250 +++++++++++++++++++------------------ 3 files changed, 202 insertions(+), 171 deletions(-) diff --git a/System/Abstracts/Pool.php b/System/Abstracts/Pool.php index b29960e9..234a1907 100644 --- a/System/Abstracts/Pool.php +++ b/System/Abstracts/Pool.php @@ -28,6 +28,36 @@ abstract class Pool extends Component public int $lastTime = 0; + protected array $hasCreate = []; + + + /** + * @param string $name + */ + public function increment(string $name) + { + if (!isset($this->hasCreate[$name])) { + $this->hasCreate[$name] = 0; + } + $this->hasCreate[$name] += 1; + } + + + /** + * @param string $name + */ + public function decrement(string $name) + { + if (!isset($this->hasCreate[$name])) { + return; + } + if ($this->hasCreate[$name] <= 0) { + return; + } + $this->hasCreate[$name] -= 1; + } + + /** * @return array * @throws ConfigException @@ -77,6 +107,18 @@ abstract class Pool extends Component } + /** + * @param $name + */ + protected function clearCreateLog($name) + { + if (!isset($this->hasCreate[$name])) { + return; + } + $this->hasCreate[$name] = 0; + } + + /** * @param $channel * @param $name @@ -90,7 +132,7 @@ abstract class Pool extends Component if ($connection) { unset($connection); } - $this->desc($name); + $this->decrement($name); } } @@ -145,6 +187,9 @@ abstract class Pool extends Component */ private function createByCallback($name, mixed $callback) { + if (!$this->canCreate($name)) { + return; + } if ($this->creates === -1 && !is_callable($callback)) { $this->creates = Timer::tick(1000, [$this, 'Heartbeat_detection']); } @@ -182,15 +227,6 @@ abstract class Pool extends Component return true; } - /** - * @param $name - * @throws Exception - */ - public function desc(string $name) - { - throw new Exception('Undefined system processing function.'); - } - /** * @param array $config @@ -204,6 +240,19 @@ abstract class Pool extends Component } + /** + * @param string $name + * @return bool + */ + public function canCreate(string $name): bool + { + if (!isset($this->hasCreate[$name])) { + $this->hasCreate[$name] = 0; + } + return $this->hasCreate[$name] >= $this->max; + } + + /** * @param $name * @return bool diff --git a/System/Pool/Connection.php b/System/Pool/Connection.php index 610fc2bb..916ffe18 100644 --- a/System/Pool/Connection.php +++ b/System/Pool/Connection.php @@ -3,11 +3,11 @@ declare(strict_types=1); namespace Snowflake\Pool; +use Exception; use HttpServer\Http\Context; use PDO; -use Exception; -use Swoole\Coroutine; use Snowflake\Abstracts\Pool; +use Swoole\Coroutine; use Swoole\Error; use Throwable; @@ -19,8 +19,6 @@ class Connection extends Pool { - public array $hasCreate = []; - public int $timeout = 1900; /** @@ -133,9 +131,6 @@ class Connection extends Pool public function get(mixed $config, $isMaster = false): mixed { $coroutineName = $this->name('mysql', $config['cds'], $isMaster); - if (!isset($this->hasCreate[$coroutineName])) { - $this->hasCreate[$coroutineName] = 0; - } if (($pdo = Context::getContext($coroutineName)) instanceof PDO) { return $pdo; } @@ -167,6 +162,9 @@ class Connection extends Pool if (!empty($charset)) { $link->query('SET NAMES ' . $charset); } + + $this->increment($name); + return $link; } @@ -178,7 +176,6 @@ class Connection extends Pool */ public function printClients($cds, $coroutineName, $isBefore = false) { -// $this->warning(($isBefore ? 'before ' : '') . 'create client[address: ' . $cds . ', ' . env('workerId') . ', coroutine: ' . Coroutine::getCid() . ', has num: ' . $this->size($coroutineName) . ', has create: ' . $this->hasCreate[$coroutineName] . ']'); } @@ -193,7 +190,7 @@ class Connection extends Pool return; } - /** @var PDO $client */ + /** @var PDO $client */ $client = Context::getContext($coroutineName); if ($client->inTransaction()) { $client->commit(); @@ -220,7 +217,7 @@ class Connection extends Pool */ public function connection_clear() { - $this->flush(0); + $this->flush(0); } @@ -243,19 +240,19 @@ class Connection extends Pool { try { if (empty($client) || !($client instanceof PDO)) { - return $result = false; + $result = false; + } else if (!$client->getAttribute(PDO::ATTR_SERVER_INFO)) { + $result = false; + } else { + $result = true; } - if (!$client->getAttribute(PDO::ATTR_SERVER_INFO)) { - return $result = false; - } - return $result = true; } catch (Error | Throwable $exception) { - $this->addError($exception, 'mysql'); - return $result = false; + $result = $this->addError($exception, 'mysql'); } finally { if (!$result) { - $this->desc($name); + $this->decrement($name); } + return $result; } } @@ -274,25 +271,4 @@ class Connection extends Pool $this->clean($coroutineName); } - /** - * @param $coroutineName - */ - public function incr($coroutineName) - { - if (!isset($this->hasCreate[$coroutineName])) { - $this->hasCreate[$coroutineName] = 0; - } - $this->hasCreate[$coroutineName] += 1; - } - - /** - * @param string $name - */ - public function desc(string $name) - { - if (!isset($this->hasCreate[$name])) { - $this->hasCreate[$name] = 0; - } - $this->hasCreate[$name] -= 1; - } } diff --git a/System/Pool/Redis.php b/System/Pool/Redis.php index 78866edb..f5ff36cc 100644 --- a/System/Pool/Redis.php +++ b/System/Pool/Redis.php @@ -5,11 +5,11 @@ declare(strict_types=1); namespace Snowflake\Pool; +use Exception; use HttpServer\Http\Context; use Redis as SRedis; -use Snowflake\Exception\RedisConnectException; -use Exception; use Snowflake\Abstracts\Pool; +use Snowflake\Exception\RedisConnectException; /** * Class RedisClient @@ -19,139 +19,145 @@ class Redis extends Pool { - public int $_create = 0; + public int $_create = 0; - /** - * @param $value - */ - public function setLength($value) - { - $this->max = $value; - } + /** + * @param $value + */ + public function setLength($value) + { + $this->max = $value; + } - /** - * @param mixed $config - * @param bool $isMaster - * @return mixed - * @throws Exception - */ - public function get(mixed $config, $isMaster = false): mixed - { - $name = $config['host'] . ':' . $config['prefix'] . ':' . $config['databases']; - $coroutineName = $this->name('redis', 'redis:' . $name, $isMaster); - if (($redis = Context::getContext($coroutineName)) instanceof \Redis) { - return $redis; - } - return Context::setContext($coroutineName, $this->getFromChannel($coroutineName, $config)); - } + /** + * @param string $name + * @return bool + */ + public function canCreate(string $name): bool + { + if (!isset($this->hasCreate[$name])) { + $this->hasCreate[$name] = 0; + } + return $this->hasCreate[$name] >= $this->max; + } - /** - * @param string $name - * @param mixed $config - * @return SRedis - * @throws RedisConnectException - * @throws Exception - */ - public function createClient(string $name, mixed $config): SRedis - { - $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'])); - } - if (empty($config['auth']) || !$redis->auth($config['auth'])) { - throw new RedisConnectException(sprintf('Redis Error: %s, Host %s, Auth %s', $redis->getLastError(), $config['host'], $config['auth'])); - } - if (!isset($config['read_timeout'])) { - $config['read_timeout'] = 10; - } - $redis->select($config['databases']); - $redis->setOption(SRedis::OPT_READ_TIMEOUT, $config['read_timeout']); - $redis->setOption(SRedis::OPT_PREFIX, $config['prefix']); - - $this->_create += 1; - - return $redis; - } + /** + * @param mixed $config + * @param bool $isMaster + * @return mixed + * @throws Exception + */ + public function get(mixed $config, $isMaster = false): mixed + { + $name = $config['host'] . ':' . $config['prefix'] . ':' . $config['databases']; + $coroutineName = $this->name('redis', 'redis:' . $name, $isMaster); + if (($redis = Context::getContext($coroutineName)) instanceof \Redis) { + return $redis; + } + return Context::setContext($coroutineName, $this->getFromChannel($coroutineName, $config)); + } - /** - * @param array $config - * @param bool $isMaster - */ - public function release(array $config, $isMaster = false) - { - $name = $config['host'] . ':' . $config['prefix'] . ':' . $config['databases']; - $coroutineName = $this->name('redis', 'redis:' . $name, $isMaster); - if (!Context::hasContext($coroutineName)) { - return; - } + /** + * @param string $name + * @param mixed $config + * @return SRedis + * @throws RedisConnectException + * @throws Exception + */ + public function createClient(string $name, mixed $config): SRedis + { + $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'])); + } + if (empty($config['auth']) || !$redis->auth($config['auth'])) { + throw new RedisConnectException(sprintf('Redis Error: %s, Host %s, Auth %s', $redis->getLastError(), $config['host'], $config['auth'])); + } + if (!isset($config['read_timeout'])) { + $config['read_timeout'] = 10; + } + $redis->select($config['databases']); + $redis->setOption(SRedis::OPT_READ_TIMEOUT, $config['read_timeout']); + $redis->setOption(SRedis::OPT_PREFIX, $config['prefix']); - $this->push($coroutineName, Context::getContext($coroutineName)); - $this->remove($coroutineName); - $this->lastTime = time(); - } + $this->increment($name); - /** - * @param array $config - * @param bool $isMaster - * @throws Exception - */ - public function destroy(array $config, $isMaster = false) - { - $name = $config['host'] . ':' . $config['prefix'] . ':' . $config['databases']; - $coroutineName = $this->name('redis', 'redis:' . $name, $isMaster); - if (Context::hasContext($coroutineName)) { - $this->desc($coroutineName); + return $redis; + } - $this->remove($coroutineName); - } - $this->flush(0); - } - /** - * @param $coroutineName - */ - public function remove(string $coroutineName) - { - Context::remove($coroutineName); - } + /** + * @param array $config + * @param bool $isMaster + */ + public function release(array $config, $isMaster = false) + { + $name = $config['host'] . ':' . $config['prefix'] . ':' . $config['databases']; + $coroutineName = $this->name('redis', 'redis:' . $name, $isMaster); + if (!Context::hasContext($coroutineName)) { + return; + } - /** - * @param string $name - * @param mixed $client - * @return bool - * @throws Exception - */ - public function checkCanUse(string $name, mixed $client): bool - { - try { - if (!($client instanceof SRedis)) { - $result = false; - } else if (!$client->isConnected() || !$client->ping('connect.')) { - $result = false; - } else { - $result = true; - } - } catch (\Throwable $exception) { - $this->addError($exception, 'redis'); - $result = false; - } finally { - if (!$result) { - $this->desc($name); - } - return $result; - } - } + $this->push($coroutineName, Context::getContext($coroutineName)); + $this->remove($coroutineName); + $this->lastTime = time(); + } + + /** + * @param array $config + * @param bool $isMaster + * @throws Exception + */ + public function destroy(array $config, $isMaster = false) + { + $name = $config['host'] . ':' . $config['prefix'] . ':' . $config['databases']; + $coroutineName = $this->name('redis', 'redis:' . $name, $isMaster); + if (Context::hasContext($coroutineName)) { + $this->decrement($coroutineName); + + $this->remove($coroutineName); + } + $this->flush(0); + } + + /** + * @param $coroutineName + */ + public function remove(string $coroutineName) + { + Context::remove($coroutineName); + } + + /** + * @param string $name + * @param mixed $client + * @return bool + * @throws Exception + */ + public function checkCanUse(string $name, mixed $client): bool + { + try { + if (!($client instanceof SRedis)) { + $result = false; + } else if (!$client->isConnected() || !$client->ping('connect.')) { + $result = false; + } else { + $result = true; + } + } catch (\Throwable $exception) { + $this->addError($exception, 'redis'); + $result = false; + } finally { + if (!$result) { + $this->decrement($name); + } + return $result; + } + } - /** - * @param string $name - */ - public function desc(string $name) - { - $this->_create -= 1; - } }