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

改名

Browse Source
This commit is contained in:
Mr·x
2021-04-23 18:37:18 +08:00
parent 9b0f919ba9
commit 8cd5562b20
3 changed files with 202 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
System/Abstracts/Pool.php
+59 -10
View File
@@ -28,6 +28,36 @@ abstract class Pool extends Component
public int $lastTime = 0; 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 * @return array
* @throws ConfigException * @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 $channel
* @param $name * @param $name
@@ -90,7 +132,7 @@ abstract class Pool extends Component
if ($connection) { if ($connection) {
unset($connection); unset($connection);
} }
$this->desc($name); $this->decrement($name);
} }
} }
@@ -145,6 +187,9 @@ abstract class Pool extends Component
*/ */
private function createByCallback($name, mixed $callback) private function createByCallback($name, mixed $callback)
{ {
if (!$this->canCreate($name)) {
return;
}
if ($this->creates === -1 && !is_callable($callback)) { if ($this->creates === -1 && !is_callable($callback)) {
$this->creates = Timer::tick(1000, [$this, 'Heartbeat_detection']); $this->creates = Timer::tick(1000, [$this, 'Heartbeat_detection']);
} }
@@ -182,15 +227,6 @@ abstract class Pool extends Component
return true; return true;
} }
/**
* @param $name
* @throws Exception
*/
public function desc(string $name)
{
throw new Exception('Undefined system processing function.');
}
/** /**
* @param array $config * @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 * @param $name
* @return bool * @return bool
System/Pool/Connection.php
+15 -39
View File
@@ -3,11 +3,11 @@ declare(strict_types=1);
namespace Snowflake\Pool; namespace Snowflake\Pool;
use Exception;
use HttpServer\Http\Context; use HttpServer\Http\Context;
use PDO; use PDO;
use Exception;
use Swoole\Coroutine;
use Snowflake\Abstracts\Pool; use Snowflake\Abstracts\Pool;
use Swoole\Coroutine;
use Swoole\Error; use Swoole\Error;
use Throwable; use Throwable;
@@ -19,8 +19,6 @@ class Connection extends Pool
{ {
public array $hasCreate = [];
public int $timeout = 1900; public int $timeout = 1900;
/** /**
@@ -133,9 +131,6 @@ class Connection extends Pool
public function get(mixed $config, $isMaster = false): mixed public function get(mixed $config, $isMaster = false): mixed
{ {
$coroutineName = $this->name('mysql', $config['cds'], $isMaster); $coroutineName = $this->name('mysql', $config['cds'], $isMaster);
if (!isset($this->hasCreate[$coroutineName])) {
$this->hasCreate[$coroutineName] = 0;
}
if (($pdo = Context::getContext($coroutineName)) instanceof PDO) { if (($pdo = Context::getContext($coroutineName)) instanceof PDO) {
return $pdo; return $pdo;
} }
@@ -167,6 +162,9 @@ class Connection extends Pool
if (!empty($charset)) { if (!empty($charset)) {
$link->query('SET NAMES ' . $charset); $link->query('SET NAMES ' . $charset);
} }
$this->increment($name);
return $link; return $link;
} }
@@ -178,7 +176,6 @@ class Connection extends Pool
*/ */
public function printClients($cds, $coroutineName, $isBefore = false) 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; return;
} }
/** @var PDO $client */ /** @var PDO $client */
$client = Context::getContext($coroutineName); $client = Context::getContext($coroutineName);
if ($client->inTransaction()) { if ($client->inTransaction()) {
$client->commit(); $client->commit();
@@ -220,7 +217,7 @@ class Connection extends Pool
*/ */
public function connection_clear() public function connection_clear()
{ {
$this->flush(0); $this->flush(0);
} }
@@ -243,19 +240,19 @@ class Connection extends Pool
{ {
try { try {
if (empty($client) || !($client instanceof PDO)) { 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) { } catch (Error | Throwable $exception) {
$this->addError($exception, 'mysql'); $result = $this->addError($exception, 'mysql');
return $result = false;
} finally { } finally {
if (!$result) { if (!$result) {
$this->desc($name); $this->decrement($name);
} }
return $result;
} }
} }
@@ -274,25 +271,4 @@ class Connection extends Pool
$this->clean($coroutineName); $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;
}
} }
System/Pool/Redis.php
+128 -122
View File
@@ -5,11 +5,11 @@ declare(strict_types=1);
namespace Snowflake\Pool; namespace Snowflake\Pool;
use Exception;
use HttpServer\Http\Context; use HttpServer\Http\Context;
use Redis as SRedis; use Redis as SRedis;
use Snowflake\Exception\RedisConnectException;
use Exception;
use Snowflake\Abstracts\Pool; use Snowflake\Abstracts\Pool;
use Snowflake\Exception\RedisConnectException;
/** /**
* Class RedisClient * Class RedisClient
@@ -19,139 +19,145 @@ class Redis extends Pool
{ {
public int $_create = 0; public int $_create = 0;
/** /**
* @param $value * @param $value
*/ */
public function setLength($value) public function setLength($value)
{ {
$this->max = $value; $this->max = $value;
} }
/** /**
* @param mixed $config * @param string $name
* @param bool $isMaster * @return bool
* @return mixed */
* @throws Exception public function canCreate(string $name): bool
*/ {
public function get(mixed $config, $isMaster = false): mixed if (!isset($this->hasCreate[$name])) {
{ $this->hasCreate[$name] = 0;
$name = $config['host'] . ':' . $config['prefix'] . ':' . $config['databases']; }
$coroutineName = $this->name('redis', 'redis:' . $name, $isMaster); return $this->hasCreate[$name] >= $this->max;
if (($redis = Context::getContext($coroutineName)) instanceof \Redis) { }
return $redis;
}
return Context::setContext($coroutineName, $this->getFromChannel($coroutineName, $config));
}
/** /**
* @param string $name * @param mixed $config
* @param mixed $config * @param bool $isMaster
* @return SRedis * @return mixed
* @throws RedisConnectException * @throws Exception
* @throws Exception */
*/ public function get(mixed $config, $isMaster = false): mixed
public function createClient(string $name, mixed $config): SRedis {
{ $name = $config['host'] . ':' . $config['prefix'] . ':' . $config['databases'];
$redis = new SRedis(); $coroutineName = $this->name('redis', 'redis:' . $name, $isMaster);
if (!$redis->pconnect($config['host'], (int)$config['port'], $config['timeout'])) { if (($redis = Context::getContext($coroutineName)) instanceof \Redis) {
throw new RedisConnectException(sprintf('The Redis Connect %s::%d Fail.', $config['host'], $config['port'])); return $redis;
} }
if (empty($config['auth']) || !$redis->auth($config['auth'])) { return Context::setContext($coroutineName, $this->getFromChannel($coroutineName, $config));
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 array $config * @param string $name
* @param bool $isMaster * @param mixed $config
*/ * @return SRedis
public function release(array $config, $isMaster = false) * @throws RedisConnectException
{ * @throws Exception
$name = $config['host'] . ':' . $config['prefix'] . ':' . $config['databases']; */
$coroutineName = $this->name('redis', 'redis:' . $name, $isMaster); public function createClient(string $name, mixed $config): SRedis
if (!Context::hasContext($coroutineName)) { {
return; $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->increment($name);
$this->remove($coroutineName);
$this->lastTime = time();
}
/** return $redis;
* @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);
$this->remove($coroutineName);
}
$this->flush(0);
}
/** /**
* @param $coroutineName * @param array $config
*/ * @param bool $isMaster
public function remove(string $coroutineName) */
{ public function release(array $config, $isMaster = false)
Context::remove($coroutineName); {
} $name = $config['host'] . ':' . $config['prefix'] . ':' . $config['databases'];
$coroutineName = $this->name('redis', 'redis:' . $name, $isMaster);
if (!Context::hasContext($coroutineName)) {
return;
}
/** $this->push($coroutineName, Context::getContext($coroutineName));
* @param string $name $this->remove($coroutineName);
* @param mixed $client $this->lastTime = time();
* @return bool }
* @throws Exception
*/ /**
public function checkCanUse(string $name, mixed $client): bool * @param array $config
{ * @param bool $isMaster
try { * @throws Exception
if (!($client instanceof SRedis)) { */
$result = false; public function destroy(array $config, $isMaster = false)
} else if (!$client->isConnected() || !$client->ping('connect.')) { {
$result = false; $name = $config['host'] . ':' . $config['prefix'] . ':' . $config['databases'];
} else { $coroutineName = $this->name('redis', 'redis:' . $name, $isMaster);
$result = true; if (Context::hasContext($coroutineName)) {
} $this->decrement($coroutineName);
} catch (\Throwable $exception) {
$this->addError($exception, 'redis'); $this->remove($coroutineName);
$result = false; }
} finally { $this->flush(0);
if (!$result) { }
$this->desc($name);
} /**
return $result; * @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;
}
} }