This commit is contained in:
2023-02-07 16:46:13 +08:00
parent a9472ff0a3
commit 4f341594b5
4 changed files with 27 additions and 46 deletions
+1 -3
View File
@@ -96,9 +96,7 @@ class Connection extends Component
*/ */
public function get(mixed $config, bool $isMaster = false): ?\PDO public function get(mixed $config, bool $isMaster = false): ?\PDO
{ {
$minx = Config::get('databases.pool.min', 1); return $this->pool->get($config['cds'] . ($isMaster ? 'master' : 'slave'), $this->generate($config));
return $this->pool->get($config['cds'] . ($isMaster ? 'master' : 'slave'), $this->generate($config), $minx);
} }
+21 -39
View File
@@ -24,9 +24,6 @@ class Pool extends Component
/** @var array<PoolQueue> */ /** @var array<PoolQueue> */
private static array $_connections = []; private static array $_connections = [];
public int $max = 60;
/** /**
* @var WorkerStatus * @var WorkerStatus
*/ */
@@ -112,18 +109,14 @@ class Pool extends Component
/** /**
* @param $name * @param $name
* @param int $max * @param int $max
* @throws ConfigException
*/ */
public function initConnections($name, int $max = 60) public function initConnections($name, int $max = 60)
{ {
if (isset(static::$_connections[$name])) { $channel = static::$_connections[$name] ?? null;
$value = static::$_connections[$name]; if (($channel instanceof PoolQueue) && !$channel->isClose()) {
if ($value instanceof PoolQueue) {
return; return;
} }
} static::$_connections[$name] = new PoolQueue($max);
$this->newChannel($name, $max);
$this->max = $max;
} }
@@ -135,37 +128,25 @@ class Pool extends Component
*/ */
public function channel($name): PoolQueue public function channel($name): PoolQueue
{ {
if (!isset(static::$_connections[$name])) { $channel = static::$_connections[$name] ?? null;
$this->newChannel($name); if (!($channel instanceof PoolQueue) ) {
throw new Exception('Channel is not exists.');
} }
if (static::$_connections[$name]->isClose()) { if ($channel->isClose()) {
throw new Exception('Channel is Close.'); throw new Exception('Channel is Close.');
} }
return static::$_connections[$name]; return $channel;
}
/**
* @throws ConfigException
*/
private function newChannel($name, $max = null)
{
if ($max == null) {
$max = Config::get('databases.pool.max', 10);
}
static::$_connections[$name] = new PoolQueue($max);
} }
/** /**
* @param $name * @param $name
* @param $callback * @param $callback
* @param $minx
* @return array * @return array
* @throws ConfigException * @throws ConfigException
* @throws Exception * @throws Exception
*/ */
public function get($name, $callback, $minx): mixed public function get($name, $callback): mixed
{ {
$channel = $this->channel($name); $channel = $this->channel($name);
if (!$channel->isEmpty()) { if (!$channel->isEmpty()) {
@@ -217,11 +198,12 @@ class Pool extends Component
*/ */
public function hasItem(string $name): bool public function hasItem(string $name): bool
{ {
if (isset(static::$_connections[$name])) { $channel = static::$_connections[$name] ?? null;
return !static::$_connections[$name]->isEmpty(); if (!($channel instanceof PoolQueue) || $channel->isClose()) {
}
return false; return false;
} }
return !$channel->isEmpty();
}
/** /**
@@ -230,10 +212,11 @@ class Pool extends Component
*/ */
public function size(string $name): int public function size(string $name): int
{ {
if (!isset(static::$_connections[$name])) { $channel = static::$_connections[$name] ?? null;
if (!($channel instanceof PoolQueue) || $channel->isClose()) {
return 0; return 0;
} }
return static::$_connections[$name]->length(); return $channel->length();
} }
@@ -248,7 +231,6 @@ class Pool extends Component
if (!$channel->isFull()) { if (!$channel->isFull()) {
$channel->push($client); $channel->push($client);
} }
unset($client);
} }
@@ -258,17 +240,17 @@ class Pool extends Component
*/ */
public function clean(string $name) public function clean(string $name)
{ {
if (!isset(static::$_connections[$name])) { $channel = static::$_connections[$name] ?? null;
if (!($channel instanceof PoolQueue) || $channel->isClose()) {
return; return;
} }
while (static::$_connections[$name]->length() > 0) { while ($channel->length() > 0) {
$client = static::$_connections[$name]->pop(); $client = $channel->pop();
if ($client instanceof StopHeartbeatCheck) { if ($client instanceof StopHeartbeatCheck) {
$client->stopHeartbeatCheck(); $client->stopHeartbeatCheck();
} }
} }
static::$_connections[$name] = null; $channel->close();
unset(static::$_connections[$name]);
} }
+2 -2
View File
@@ -82,7 +82,7 @@ class Helper implements StopHeartbeatCheck
/** /**
* @return \Redis * @return \Redis
* @throws RedisConnectException * @throws Exception
* @throws RedisException * @throws RedisException
*/ */
public function _pdo(): \Redis public function _pdo(): \Redis
@@ -96,7 +96,7 @@ class Helper implements StopHeartbeatCheck
/** /**
* @return \Redis * @return \Redis
* @throws RedisConnectException * @throws Exception
*/ */
private function newClient(): \Redis private function newClient(): \Redis
{ {
+2 -1
View File
@@ -91,6 +91,7 @@ class Redis extends Component
* @param $key * @param $key
* @param int $timeout * @param int $timeout
* @return bool * @return bool
* @throws \RedisException
*/ */
public function waite($key, int $timeout = 5): bool public function waite($key, int $timeout = 5): bool
{ {
@@ -189,7 +190,7 @@ SCRIPT;
$config = $this->get_config(); $config = $this->get_config();
return $this->pool->get($config['host'], static function () use ($config) { return $this->pool->get($config['host'], static function () use ($config) {
return new Helper($config); return new Helper($config);
}, 10); });
} }