as2252258/kiri-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

111

Browse Source
This commit is contained in:
as2252258@163.com
2021-09-06 01:36:03 +08:00
parent 0116fb2e51
commit 896f4e08f6
1 changed files with 182 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/Pool/Pool.php
+182 -178
View File
@@ -20,214 +20,218 @@ use Swoole\Coroutine\Channel;
class Pool extends Component class Pool extends Component
{ {
/** @var Channel[] */ /** @var Channel[] */
private static array $_connections = []; private static array $_connections = [];
public int $max = 60; public int $max = 60;
use Alias; use Alias;
/** /**
* @param $channel * @param $channel
* @param $retain_number * @param $retain_number
* @throws Exception * @throws Exception
*/ */
public function flush($channel, $retain_number) public function flush($channel, $retain_number)
{ {
$this->pop($channel, $retain_number); $this->pop($channel, $retain_number);
} }
/** /**
* @param Channel $channel * @param Channel $channel
* @param $retain_number * @param $retain_number
* @throws Exception * @throws Exception
*/ */
protected function pop(Channel $channel, $retain_number): void protected function pop(Channel $channel, $retain_number): void
{ {
while ($channel->length() > $retain_number) { if ($channel instanceof Channel && Coroutine::getCid() === -1) {
$connection = $channel->pop(); unset($channel);
if ($connection instanceof StopHeartbeatCheck) { return;
$connection->stopHeartbeatCheck(); }
} while ($channel->length() > $retain_number) {
} $connection = $channel->pop();
} if ($connection instanceof StopHeartbeatCheck) {
$connection->stopHeartbeatCheck();
}
}
}
/** /**
* @param $name * @param $name
* @param false $isMaster * @param false $isMaster
* @param int $max * @param int $max
* @throws ConfigException * @throws ConfigException
*/ */
public function initConnections($name, bool $isMaster = false, int $max = 60) public function initConnections($name, bool $isMaster = false, int $max = 60)
{ {
$name = $this->name($name, $isMaster); $name = $this->name($name, $isMaster);
if (isset(static::$_connections[$name])) { if (isset(static::$_connections[$name])) {
$value = static::$_connections[$name]; $value = static::$_connections[$name];
if ($value instanceof Channel || $value instanceof SplQueue) { if ($value instanceof Channel || $value instanceof SplQueue) {
return; return;
} }
} }
$this->newChannel($name, $max); $this->newChannel($name, $max);
$this->max = $max; $this->max = $max;
} }
/** /**
* @param $name * @param $name
* @return Channel|SplQueue * @return Channel|SplQueue
* @throws ConfigException * @throws ConfigException
* @throws Exception * @throws Exception
*/ */
private function getChannel($name): Channel|SplQueue private function getChannel($name): Channel|SplQueue
{ {
if (!isset(static::$_connections[$name])) { if (!isset(static::$_connections[$name])) {
$this->newChannel($name); $this->newChannel($name);
} }
if (static::$_connections[$name]->errCode == SWOOLE_CHANNEL_CLOSED) { if (static::$_connections[$name]->errCode == SWOOLE_CHANNEL_CLOSED) {
throw new Exception('Channel is Close.'); throw new Exception('Channel is Close.');
} }
return static::$_connections[$name]; return static::$_connections[$name];
} }
/** /**
* @throws ConfigException * @throws ConfigException
*/ */
private function newChannel($name, $max = null) private function newChannel($name, $max = null)
{ {
if ($max == null) { if ($max == null) {
$max = Config::get('databases.pool.max', 10); $max = Config::get('databases.pool.max', 10);
} }
if (Coroutine::getCid() === -1) { if (Coroutine::getCid() === -1) {
static::$_connections[$name] = new SplQueue($max); static::$_connections[$name] = new SplQueue($max);
} else { } else {
static::$_connections[$name] = new Channel($max); static::$_connections[$name] = new Channel($max);
} }
} }
/** /**
* @param $name * @param $name
* @param $callback * @param $callback
* @return array * @return array
* @throws ConfigException * @throws ConfigException
*/ */
public function get($name, $callback): mixed public function get($name, $callback): mixed
{ {
$channel = $this->getChannel($name); $channel = $this->getChannel($name);
if (!$channel->isEmpty()) { if (!$channel->isEmpty()) {
$connection = $channel->pop(); $connection = $channel->pop();
defer(fn() => $this->maxIdleQuantity($channel)); defer(fn() => $this->maxIdleQuantity($channel));
if ($this->checkCanUse($name, $connection)) { if ($this->checkCanUse($name, $connection)) {
return $connection; return $connection;
} }
} }
return $callback(); return $callback();
} }
/** /**
* @param $channel * @param $channel
* @throws ConfigException * @throws ConfigException
* @throws Exception * @throws Exception
*/ */
private function maxIdleQuantity($channel): void private function maxIdleQuantity($channel): void
{ {
$minx = Config::get('databases.pool.min', 1); $minx = Config::get('databases.pool.min', 1);
if ($channel->length() > $minx) { if ($channel->length() > $minx) {
$this->pop($channel, $minx); $this->pop($channel, $minx);
} }
} }
/** /**
* @param $name * @param $name
* @return bool * @return bool
* @throws ConfigException * @throws ConfigException
*/ */
public function isNull($name): bool public function isNull($name): bool
{ {
return $this->getChannel($name)->isEmpty(); return $this->getChannel($name)->isEmpty();
} }
/** /**
* @param string $name * @param string $name
* @param mixed $client * @param mixed $client
* @return bool * @return bool
* 检查连接可靠性 * 检查连接可靠性
*/ */
public function checkCanUse(string $name, mixed $client): bool public function checkCanUse(string $name, mixed $client): bool
{ {
return true; return true;
} }
/** /**
* @param string $name * @param string $name
* @return bool * @return bool
*/ */
public function hasItem(string $name): bool public function hasItem(string $name): bool
{ {
if (isset(static::$_connections[$name])) { if (isset(static::$_connections[$name])) {
return !static::$_connections[$name]->isEmpty(); return !static::$_connections[$name]->isEmpty();
} }
return false; return false;
} }
/** /**
* @param string $name * @param string $name
* @return mixed * @return mixed
*/ */
public function size(string $name): mixed public function size(string $name): mixed
{ {
if (!isset(static::$_connections[$name])) { if (!isset(static::$_connections[$name])) {
return 0; return 0;
} }
return static::$_connections[$name]->length(); return static::$_connections[$name]->length();
} }
/** /**
* @param string $name * @param string $name
* @param mixed $client * @param mixed $client
* @throws ConfigException * @throws ConfigException
*/ */
public function push(string $name, mixed $client) public function push(string $name, mixed $client)
{ {
$channel = $this->getChannel($name); $channel = $this->getChannel($name);
if (!$channel->isFull()) { if (!$channel->isFull()) {
$channel->push($client); $channel->push($client);
} }
unset($client); unset($client);
} }
/** /**
* @param string $name * @param string $name
* @throws Exception * @throws Exception
*/ */
public function clean(string $name) public function clean(string $name)
{ {
if (!isset(static::$_connections[$name])) { if (!isset(static::$_connections[$name])) {
return; return;
} }
$channel = static::$_connections[$name]; $channel = static::$_connections[$name];
$this->pop($channel, 0); $this->pop($channel, 0);
} }
/** /**
* @return Channel[] * @return Channel[]
*/ */
protected function getChannels(): array protected function getChannels(): array
{ {
return static::$_connections; return static::$_connections;
} }
} }