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

modify

Browse Source
This commit is contained in:
as2252258@163.com
2021-02-22 00:56:05 +08:00
parent 8a40bd7130
commit 985157ad8b
5 changed files with 723 additions and 768 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Database/Db.php
+2 -2
View File
@@ -49,7 +49,7 @@ class Db
$event = Snowflake::app()->getEvent(); $event = Snowflake::app()->getEvent();
$event->trigger(Connection::TRANSACTION_COMMIT); $event->trigger(Connection::TRANSACTION_COMMIT);
$event->offName(Connection::TRANSACTION_COMMIT); $event->offName(Connection::TRANSACTION_COMMIT);
Context::deleteId('begin:transaction'); Context::deleteContext('begin:transaction');
} }
/** /**
@@ -63,7 +63,7 @@ class Db
$event = Snowflake::app()->getEvent(); $event = Snowflake::app()->getEvent();
$event->trigger(Connection::TRANSACTION_ROLLBACK); $event->trigger(Connection::TRANSACTION_ROLLBACK);
$event->offName(Connection::TRANSACTION_ROLLBACK); $event->offName(Connection::TRANSACTION_ROLLBACK);
Context::deleteId('begin:transaction'); Context::deleteContext('begin:transaction');
} }
/** /**
HttpServer/Http/Context.php
+1 -1
View File
@@ -149,7 +149,7 @@ class Context extends BaseContext
* @param string $id * @param string $id
* @param string|null $key * @param string|null $key
*/ */
public static function deleteId(string $id, string $key = null) public static function deleteContext(string $id, string $key = null)
{ {
if (!static::hasContext($id, $key)) { if (!static::hasContext($id, $key)) {
return; return;
System/Abstracts/Pool.php
+31 -36
View File
@@ -54,23 +54,43 @@ abstract class Pool extends Component
* @return array * @return array
* @throws Exception * @throws Exception
*/ */
protected function get($name): array protected function get($name, array $callback): mixed
{ {
if (!Context::inCoroutine()) { if (!Context::inCoroutine()) {
return [0, null]; return $this->createClient($name, $callback);
} }
[$timeout, $connection] = $this->_items[$name]->pop(-1); if (!$this->hasItem($name)) {
if (empty($timeout) || empty($connection)) { if ($this->creates === -1) {
return [0, null]; $this->creates = Timer::tick(1000, [$this, 'Heartbeat_detection']);
} }
if (!$this->checkCanUse($name, $timeout, $connection)) { if (!Context::hasContext('create::client::ing::' . $name)) {
return [0, null]; $this->push($name, $this->createClient($name, $callback));
Context::deleteContext('create::client::ing::' . $name);
}
}
$connection = $this->_items[$name]->pop(-1);
if (!$this->checkCanUse($name, $connection)) {
return $this->createClient($name, $callback);
} else { } else {
return [$timeout, $connection]; return $connection;
} }
} }
/**
* @param $cds
* @param $coroutineName
* @param false $isBefore
*/
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->_create . ']');
}
abstract public function createClient(string $name, array $config): mixed;
/** /**
* @param $driver * @param $driver
* @param $cds * @param $cds
@@ -95,7 +115,7 @@ abstract class Pool extends Component
* 检查连接可靠性 * 检查连接可靠性
* @throws Exception * @throws Exception
*/ */
public function checkCanUse(string $name, int $time, mixed $client): mixed public function checkCanUse(string $name, mixed $client): mixed
{ {
throw new Exception('Undefined system processing function.'); throw new Exception('Undefined system processing function.');
} }
@@ -120,30 +140,6 @@ abstract class Pool extends Component
} }
/**
* @param array $config
* @param string $coroutineName
* @param callable $createHandler
* @throws Exception
*/
public function createConnect(array $config, string $coroutineName, callable $createHandler)
{
if (Context::hasContext('create:connect:' . $coroutineName)) {
return;
}
if ($this->creates === -1) {
$this->creates = Timer::tick(1000, [$this, 'Heartbeat_detection']);
}
Context::setContext('create:connect:' . $coroutineName, 1);
$this->push($coroutineName, call_user_func($createHandler, ...$config));
Context::deleteId('create:connect:' . $coroutineName);
}
/** /**
* @param $name * @param $name
* @return bool * @return bool
@@ -183,7 +179,7 @@ abstract class Pool extends Component
return; return;
} }
if (!$this->_items[$name]->isFull()) { if (!$this->_items[$name]->isFull()) {
$this->_items[$name]->push([time(), $client]); $this->_items[$name]->push($client);
} }
} }
@@ -201,9 +197,8 @@ abstract class Pool extends Component
return; return;
} }
$channel = $this->_items[$name]; $channel = $this->_items[$name];
while ([$time, $client] = $channel->pop(0.001)) { while ($client = $channel->pop(0.001)) {
unset($client); unset($client);
$this->desc($name); $this->desc($name);
} }
} }
System/Pool/Connection.php
+13 -32
View File
@@ -143,27 +143,23 @@ class Connection extends Pool
if (($pdo = Context::getContext($coroutineName)) instanceof PDO) { if (($pdo = Context::getContext($coroutineName)) instanceof PDO) {
return $pdo; return $pdo;
} }
if (!$this->hasItem($coroutineName)) { $connections = $this->get($coroutineName, $config);
$this->newClient($config, $coroutineName); if ($number = Context::getContext('begin_' . $coroutineName, Coroutine::getCid())) {
} $number > 0 && $connections->beginTransaction();
[$time, $connections] = $this->get($coroutineName);
if (!($connections instanceof PDO)) {
return $this->getConnection($config, $coroutineName);
} }
return Context::setContext($coroutineName, $connections); return Context::setContext($coroutineName, $connections);
} }
/** /**
* @param $config * @param array $config
* @param $coroutineName * @return mixed
* @throws Exception
*/ */
private function newClient($config, $coroutineName) public function createClient(string $name, array $config): mixed
{ {
$this->printClients($config['cds'], $coroutineName, true); $this->printClients($config['cds'], $name, true);
$this->createConnect($this->parseConfig($config, $coroutineName), $coroutineName, function ($cds, $username, $password, $charset, $coroutineName) { // TODO: Implement createClient() method.
$link = new PDO($cds, $username, $password, [ $link = new PDO($config['cds'], $config['username'], $config['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,
@@ -171,26 +167,12 @@ class Connection extends Pool
$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);
$charset = $config['charset'] ?? 'utf8mb4';
if (!empty($charset)) { if (!empty($charset)) {
$link->query('SET NAMES ' . $charset); $link->query('SET NAMES ' . $charset);
} }
$this->incr($coroutineName);
if ($number = Context::getContext('begin_' . $coroutineName, Coroutine::getCid())) {
$number > 0 && $link->beginTransaction();
}
return $link; return $link;
});
}
/**
* @param $config
* @param $name
* @return array
*/
private function parseConfig($config, $name): array
{
return [$config['cds'], $config['username'], $config['password'], $config['charset'] ?? 'utf8mb4', $name];
} }
@@ -264,16 +246,15 @@ class Connection extends Pool
*/ */
public function remove($coroutineName) public function remove($coroutineName)
{ {
Context::deleteId($coroutineName); Context::deleteContext($coroutineName);
} }
/** /**
* @param $name * @param $name
* @param $time
* @param $client * @param $client
* @return bool * @return bool
*/ */
public function checkCanUse($name, $time, $client): bool public function checkCanUse($name, $client): bool
{ {
try { try {
if (empty($client) || !($client instanceof PDO)) { if (empty($client) || !($client instanceof PDO)) {
System/Pool/Redis.php
+8 -29
View File
@@ -21,7 +21,6 @@ class Redis extends Pool
{ {
private int $_create = 0; private int $_create = 0;
/** /**
@@ -46,26 +45,18 @@ class Redis extends Pool
if (($redis = Context::getContext($coroutineName)) instanceof \Redis) { if (($redis = Context::getContext($coroutineName)) instanceof \Redis) {
return $redis; return $redis;
} }
if (!$this->hasItem($coroutineName)) { return Context::setContext($coroutineName, $this->get($coroutineName, $config));
$this->newClient($config, $coroutineName);
}
[$time, $clients] = $this->get($coroutineName);
if ($clients === null) {
return $this->getConnection($config, $coroutineName);
}
return Context::setContext($coroutineName, $clients);
} }
/** /**
* @param $config * @param array $config
* @param $coroutineName * @return SRedis
* @throws Exception * @throws RedisConnectException
*/ */
private function newClient($config, $coroutineName) public function createClient(string $name, array $config): mixed
{ {
$this->printClients($config['host'], $coroutineName, true); $this->printClients($config['host'], $name, true);
$this->createConnect([$config, $coroutineName], $coroutineName, function ($config, $coroutineName) {
$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']));
@@ -83,18 +74,6 @@ class Redis extends Pool
$this->_create += 1; $this->_create += 1;
return $redis; return $redis;
});
}
/**
* @param $cds
* @param $coroutineName
* @param false $isBefore
*/
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->_create . ']');
} }
@@ -138,7 +117,7 @@ class Redis extends Pool
*/ */
public function remove(string $coroutineName) public function remove(string $coroutineName)
{ {
Context::deleteId($coroutineName); Context::deleteContext($coroutineName);
} }
/** /**
@@ -148,7 +127,7 @@ class Redis extends Pool
* @return bool * @return bool
* @throws Exception * @throws Exception
*/ */
public function checkCanUse(string $name, int $time, mixed $client): bool public function checkCanUse(string $name, mixed $client): bool
{ {
try { try {
if (!($client instanceof SRedis)) { if (!($client instanceof SRedis)) {