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-01-04 18:40:11 +08:00
parent b4f6d1be35
commit 0c7a777ab4
3 changed files with 148 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
System/Abstracts/Pool.php
+22
View File
@@ -6,6 +6,7 @@ namespace Snowflake\Abstracts;
use Exception; use Exception;
use HttpServer\Http\Context;
use Swoole\Coroutine; use Swoole\Coroutine;
use Swoole\Coroutine\Channel; use Swoole\Coroutine\Channel;
@@ -99,6 +100,27 @@ abstract class Pool extends Component
throw new Exception('Undefined system processing function.'); throw new Exception('Undefined system processing function.');
} }
/**
* @param array $config
* @param string $coroutineName
* @param callable $createHandler
*/
public function createConnect(array $config, string $coroutineName, callable $createHandler): void
{
if (Context::hasContext('create:connect:' . $coroutineName)) {
return;
}
Context::setContext('create:connect:' . $coroutineName, 1);
$client = call_user_func($createHandler, ...$config);
$this->push($coroutineName, $client);
Context::deleteId('create:connect:' . $coroutineName);
}
/** /**
* @param $name * @param $name
* @return bool * @return bool
System/Pool/Connection.php
+63 -58
View File
@@ -207,45 +207,44 @@ class Connection extends Pool
return Context::getContext($coroutineName); return Context::getContext($coroutineName);
} }
if ($this->size($coroutineName) < 1 && $this->hasCreate[$coroutineName] < $this->max) { if ($this->size($coroutineName) < 1 && $this->hasCreate[$coroutineName] < $this->max) {
return $this->saveClient($coroutineName, $this->nowClient($coroutineName, $config)); $this->newClient($config, $coroutineName);
} }
$connections = $client = $this->get($coroutineName); [$time, $connections] = $this->get($coroutineName);
if ($connections[1] instanceof PDO) { if ($connections instanceof PDO) {
return $this->saveClient($coroutineName, $connections[1]); throw new Exception('Database exception.');
} }
return $this->saveClient($coroutineName, $this->nowClient($coroutineName, $config)); return Context::setContext($coroutineName, $connections);
} }
/** /**
* @param $coroutineName
* @param $client
* @return mixed
*/
private function saveClient($coroutineName, $client): mixed
{
return Context::setContext($coroutineName, $client);
}
/**
* @param $coroutineName
* @param $config * @param $config
* @return PDO * @param $coroutineName
* @throws Exception
*/ */
private function nowClient($coroutineName, $config): PDO private function newClient($config, $coroutineName)
{ {
$this->createConnect($this->parseConfig($config, $coroutineName), $coroutineName, function ($cds, $username, $password, $charset, $coroutineName) {
$this->createConnect($coroutineName, ...$this->parseConfig($config)); $link = new PDO($cds, $username, $password, [
[$time, $client] = $this->get($coroutineName); PDO::ATTR_EMULATE_PREPARES => false,
PDO::ATTR_CASE => PDO::CASE_NATURAL,
PDO::ATTR_TIMEOUT => $this->timeout,
]);
$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);
if (!empty($charset)) {
$link->query('SET NAMES ' . $charset);
}
$this->success('create db client -> ' . $cds . ':' . $this->hasCreate[$coroutineName] . ':' . $this->size($coroutineName));
$this->incr($coroutineName);
if ($number = Context::getContext('begin_' . $coroutineName, Coroutine::getCid())) { if ($number = Context::getContext('begin_' . $coroutineName, Coroutine::getCid())) {
$number > 0 && $client->beginTransaction(); $number > 0 && $link->beginTransaction();
} }
if ($this->creates === 0) { if ($this->creates === 0) {
$this->creates = Timer::tick(10000, [$this, 'Heartbeat_detection']); $this->creates = Timer::tick(10000, [$this, 'Heartbeat_detection']);
} }
return $client; return $link;
});
} }
@@ -253,9 +252,9 @@ class Connection extends Pool
* @param $config * @param $config
* @return array * @return array
*/ */
private function parseConfig($config): array private function parseConfig($config, $name): array
{ {
return [$config['cds'], $config['username'], $config['password'], $config['charset'] ?? 'utf8mb4']; return [$config['cds'], $config['username'], $config['password'], $config['charset'] ?? 'utf8mb4', $name];
} }
@@ -359,37 +358,43 @@ class Connection extends Pool
* @param string $charset * @param string $charset
* @throws Exception * @throws Exception
*/ */
public function createConnect($coroutineName, $cds, $username, $password, $charset = 'utf8mb4'): void // public function createConnect($coroutineName, $cds, $username, $password, $charset = 'utf8mb4'): void
{ // {
try { // try {
if (Context::hasContext('at_create_db')) { // if (Context::hasContext('at_create_db')) {
return; // return;
} // }
Context::setContext('at_create_db',1); // Context::setContext('at_create_db', 1);
$link = new PDO($cds, $username, $password, [ // $link = new PDO($cds, $username, $password, [
PDO::ATTR_EMULATE_PREPARES => false, // PDO::ATTR_EMULATE_PREPARES => false,
PDO::ATTR_CASE => PDO::CASE_NATURAL, // PDO::ATTR_CASE => PDO::CASE_NATURAL,
PDO::ATTR_TIMEOUT => $this->timeout, // PDO::ATTR_TIMEOUT => $this->timeout,
]); // ]);
$link->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); // $link->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$link->setAttribute(PDO::ATTR_STRINGIFY_FETCHES, false); // $link->setAttribute(PDO::ATTR_STRINGIFY_FETCHES, false);
$link->setAttribute(PDO::ATTR_ORACLE_NULLS, PDO::NULL_EMPTY_STRING); // $link->setAttribute(PDO::ATTR_ORACLE_NULLS, PDO::NULL_EMPTY_STRING);
if (!empty($charset)) { // if (!empty($charset)) {
$link->query('SET NAMES ' . $charset); // $link->query('SET NAMES ' . $charset);
} // }
$this->success('create db client -> ' . $cds . ':' . $this->hasCreate[$coroutineName] . ':' . $this->size($coroutineName)); // $this->success('create db client -> ' . $cds . ':' . $this->hasCreate[$coroutineName] . ':' . $this->size($coroutineName));
$this->incr($coroutineName); // $this->incr($coroutineName);
$this->push($coroutineName, $link); // if ($number = Context::getContext('begin_' . $coroutineName, Coroutine::getCid())) {
} catch (\Throwable $exception) { // $number > 0 && $link->beginTransaction();
$this->addError($cds . ' -> ' . $exception->getMessage()); // }
if ($exception->getCode() === 2006) { // if ($this->creates === 0) {
$this->createConnect($coroutineName, $cds, $username, $password, $charset); // $this->creates = Timer::tick(10000, [$this, 'Heartbeat_detection']);
} // }
throw new Exception($exception->getMessage()); // $this->push($coroutineName, $link);
} finally { // } catch (\Throwable $exception) {
Context::deleteId('at_create_db'); // $this->addError($cds . ' -> ' . $exception->getMessage());
} // if ($exception->getCode() === 2006) {
} // $this->createConnect($coroutineName, $cds, $username, $password, $charset);
// }
// throw new Exception($exception->getMessage());
// } finally {
// Context::deleteId('at_create_db');
// }
// }
/** /**
* @param $coroutineName * @param $coroutineName
System/Pool/Redis.php
+58 -35
View File
@@ -58,7 +58,7 @@ class Redis extends Pool
public function getByChannel($coroutineName, $config): mixed public function getByChannel($coroutineName, $config): mixed
{ {
if ($this->_create < $this->max && !$this->hasItem($coroutineName)) { if ($this->_create < $this->max && !$this->hasItem($coroutineName)) {
$this->createConnect($config, $coroutineName); $this->newClient($config, $coroutineName);
} }
[$time, $clients] = $this->get($coroutineName); [$time, $clients] = $this->get($coroutineName);
if ($clients === null) { if ($clients === null) {
@@ -68,6 +68,29 @@ class Redis extends Pool
} }
private function newClient($config, $coroutineName)
{
$this->createConnect([$config, $coroutineName], $coroutineName, function ($config, $coroutineName) {
$redis = new SRedis();
if (!$redis->connect($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']));
}
$this->success('create redis client -> ' . $config['host'] . ':' . $this->size($coroutineName));
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']);
return $redis;
});
}
/** /**
* @param $coroutineName * @param $coroutineName
* @param $client * @param $client
@@ -85,40 +108,40 @@ class Redis extends Pool
* @param string $coroutineName * @param string $coroutineName
* @throws Exception * @throws Exception
*/ */
private function createConnect(array $config, string $coroutineName): void // private function createConnect(array $config, string $coroutineName): void
{ // {
try { // try {
if ($this->_create >= $this->max) { // if ($this->_create >= $this->max) {
return; // return;
} // }
$this->_create += 1; // $this->_create += 1;
if (Context::hasContext('at_create_db')) { // if (Context::hasContext('at_create_db')) {
return; // return;
} // }
Context::setContext('at_create_db', 1); // Context::setContext('at_create_db', 1);
$redis = new SRedis(); // $redis = new SRedis();
if (!$redis->connect($config['host'], (int)$config['port'], $config['timeout'])) { // if (!$redis->connect($config['host'], (int)$config['port'], $config['timeout'])) {
throw new RedisConnectException(sprintf('The Redis Connect %s::%d Fail.', $config['host'], $config['port'])); // throw new RedisConnectException(sprintf('The Redis Connect %s::%d Fail.', $config['host'], $config['port']));
} // }
if (empty($config['auth']) || !$redis->auth($config['auth'])) { // 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'])); // throw new RedisConnectException(sprintf('Redis Error: %s, Host %s, Auth %s', $redis->getLastError(), $config['host'], $config['auth']));
} // }
$this->success('create redis client -> ' . $config['host'] . ':' . $this->size($coroutineName)); // $this->success('create redis client -> ' . $config['host'] . ':' . $this->size($coroutineName));
if (!isset($config['read_timeout'])) { // if (!isset($config['read_timeout'])) {
$config['read_timeout'] = 10; // $config['read_timeout'] = 10;
} // }
$redis->select($config['databases']); // $redis->select($config['databases']);
$redis->setOption(SRedis::OPT_READ_TIMEOUT, $config['read_timeout']); // $redis->setOption(SRedis::OPT_READ_TIMEOUT, $config['read_timeout']);
$redis->setOption(SRedis::OPT_PREFIX, $config['prefix']); // $redis->setOption(SRedis::OPT_PREFIX, $config['prefix']);
//
$this->push($coroutineName, $redis); // $this->push($coroutineName, $redis);
} catch (\Throwable $exception) { // } catch (\Throwable $exception) {
$this->addError($exception->getMessage()); // $this->addError($exception->getMessage());
} finally { // } finally {
Context::deleteId('at_create_db'); // Context::deleteId('at_create_db');
} // }
//
} // }
/** /**
* @param array $config * @param array $config