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-07-12 17:45:04 +08:00
parent 25c00ea088
commit 4736478ddf
7 changed files with 321 additions and 689 deletions
Download Patch File Download Diff File Expand all files Collapse all files
System/Abstracts/BaseApplication.php
+4 -4
View File
@@ -38,7 +38,7 @@ use Snowflake\Event;
use Snowflake\Exception\InitException;
use Snowflake\Exception\NotFindClassException;
use Snowflake\Jwt\Jwt;
use Snowflake\Pool\ClientsPool;
use Snowflake\Pool\Pool;
use Snowflake\Pool\Connection;
use Snowflake\Pool\Redis as SRedis;
use Snowflake\Snowflake;
@@ -435,10 +435,10 @@ abstract class BaseApplication extends Service
/**
* @return ClientsPool
* @return Pool
* @throws Exception
*/
public function getClientsPool(): ClientsPool
public function getClientsPool(): Pool
{
return $this->get('clientsPool');
}
@@ -453,7 +453,7 @@ abstract class BaseApplication extends Service
'error' => ['class' => ErrorHandler::class],
'connections' => ['class' => Connection::class],
'redis_connections' => ['class' => SRedis::class],
'clientsPool' => ['class' => ClientsPool::class],
'clientsPool' => ['class' => Pool::class],
'config' => ['class' => Config::class],
'logger' => ['class' => Logger::class],
'annotation' => ['class' => SAnnotation::class],
System/Abstracts/Pool.php
-328
View File
@@ -1,328 +0,0 @@
<?php
declare(strict_types=1);
namespace Snowflake\Abstracts;
use Exception;
use JetBrains\PhpStorm\Pure;
use Snowflake\Exception\ConfigException;
use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use Swoole\Timer;
/**
* Class Pool
* @package Snowflake\Snowflake\Pool
*/
abstract class Pool extends Component
{
/** @var Channel[] */
private static array $_items = [];
public int $max = 60;
public int $creates = -1;
public int $lastTime = 0;
protected static array $hasCreate = [];
/**
* @param string $name
*/
public function increment(string $name)
{
if (!isset(static::$hasCreate[$name])) {
static::$hasCreate[$name] = 0;
}
static::$hasCreate[$name] += 1;
}
/**
* @param string $name
*/
public function decrement(string $name)
{
if (!isset(static::$hasCreate[$name])) {
return;
}
if (static::$hasCreate[$name] <= 0) {
return;
}
static::$hasCreate[$name] -= 1;
}
/**
* @return array
* @throws ConfigException
*/
private function getClearTime(): array
{
$firstClear = Config::get('pool.clear.start', 600);
$lastClear = Config::get('pool.clear.end', 300);
return [$firstClear, $lastClear];
}
/**
* @throws Exception
*/
public function Heartbeat_detection($ticker, string $name)
{
if (env('state') == 'exit') {
Timer::clear($this->creates);
$this->creates = -1;
} else {
$min = Config::get('databases.pool.min', 1);
if (($length = $this->getChannel($name)->length()) > $min) {
$this->flush($min);
}
}
}
/**
* @param $retain_number
* @throws Exception
*/
protected function flush($retain_number)
{
$channels = $this->getChannels();
foreach ($channels as $name => $channel) {
$names[] = $name;
$this->pop($channel, $name, $retain_number);
}
static::$_items = [];
if ($retain_number == 0) {
Timer::clear($this->creates);
$this->creates = -1;
}
}
/**
* @param $name
*/
protected function clearCreateLog($name): void
{
if (!isset(static::$hasCreate[$name])) {
return;
}
static::$hasCreate[$name] = 0;
}
/**
* @param Channel $channel
* @param $name
* @param $retain_number
* @throws Exception
*/
protected function pop(Channel $channel, $name, $retain_number): void
{
if (Coroutine::getCid() === -1) {
return;
}
while ($channel->length() > $retain_number) {
$connection = $channel->pop();
if ($connection) {
unset($connection);
}
$this->decrement($name);
}
}
/**
* @param $driver
* @param $name
* @param false $isMaster
* @param int $max
*/
public function initConnections($driver, $name, bool $isMaster = false, int $max = 60)
{
$name = $this->name($driver, $name, $isMaster);
if (isset(static::$_items[$name]) && static::$_items[$name] instanceof Channel) {
return;
}
if (Coroutine::getCid() === -1) {
return;
}
if ($this->creates === -1) {
$this->creates = Timer::tick(1000, [$this, 'Heartbeat_detection'], $name);
}
static::$_items[$name] = new Channel($max);
$this->max = $max;
}
/**
* @param $name
* @return Channel
* @throws ConfigException
*/
private function getChannel($name): Channel
{
if (!isset(static::$_items[$name])) {
static::$_items[$name] = new Channel(Config::get('databases.pool.max', 10));
if ($this->creates === -1) {
$this->creates = Timer::tick(1000, [$this, 'Heartbeat_detection'], $name);
}
}
return static::$_items[$name];
}
/**
* @param $name
* @param array $callback
* @return array
* @throws Exception
*/
protected function getFromChannel($name, mixed $callback): mixed
{
if (Coroutine::getCid() === -1) {
return $this->createClient($name, $callback);
}
$channel = $this->getChannel($name);
if (!$channel->isEmpty()) {
$connection = $channel->pop();
if ($this->checkCanUse($name, $connection)) {
return $connection;
}
}
return $this->createClient($name, $callback);
}
/**
* @param string $name
* @param mixed $config
* @return mixed
*/
abstract public function createClient(string $name, mixed $config): mixed;
/**
* @param $driver
* @param $cds
* @param false $isMaster
* @return string
*/
#[Pure] public function name($driver, $cds, bool $isMaster = false): string
{
if ($isMaster === true) {
return $cds . '_master';
} else {
return $cds . '_slave';
}
}
/**
* @param string $name
* @param mixed $client
* @return bool
* 检查连接可靠性
*/
public function checkCanUse(string $name, mixed $client): bool
{
return true;
}
/**
* @param array $config
* @param bool $isMaster
* @return mixed
* @throws Exception
*/
public function get(mixed $config, bool $isMaster): mixed
{
throw new Exception('Undefined system processing function.');
}
/**
* @param string $name
* @return bool
*/
public function hasItem(string $name): bool
{
if (isset(static::$_items[$name])) {
return !static::$_items[$name]->isEmpty();
}
return false;
}
/**
* @param string $name
* @return mixed
*/
public function size(string $name): mixed
{
if (Coroutine::getCid() === -1) {
return 0;
}
if (!isset(static::$_items[$name])) {
return 0;
}
return static::$_items[$name]->length();
}
/**
* @param string $name
* @param mixed $client
* @throws ConfigException
*/
public function push(string $name, mixed $client)
{
if (Coroutine::getCid() === -1) {
return;
}
$channel = $this->getChannel($name);
if (!$channel->isFull()) {
$channel->push($client);
}
unset($client);
}
/**
* @param string $name
* @throws Exception
*/
public function clean(string $name)
{
if (Coroutine::getCid() === -1 || !isset(static::$_items[$name])) {
return;
}
$channel = static::$_items[$name];
$this->pop($channel, $name, 0);
if ($this->creates > -1) {
Timer::clear($this->creates);
}
$channel->close();
static::$_items[$name] = null;
}
/**
* @return Channel[]
*/
protected function getChannels(): array
{
return static::$_items;
}
}
System/Abstracts/TraitApplication.php
+2 -2
View File
@@ -23,7 +23,7 @@ use Snowflake\Channel;
use Snowflake\Error\Logger;
use Snowflake\Event;
use Snowflake\Jwt\Jwt;
use Snowflake\Pool\ClientsPool;
use Snowflake\Pool\Pool;
use Snowflake\Pool\Connection;
use Rpc\Producer as RPCProducer;
@@ -53,7 +53,7 @@ use Rpc\Producer as RPCProducer;
* @property Channel $channel
* @property Shutdown $shutdown
* @property Emit $emit
* @property ClientsPool $clientsPool
* @property Pool $clientsPool
*/
trait TraitApplication
{
System/Pool/ClientsPool.php
-332
View File
@@ -1,332 +0,0 @@
<?php
namespace Snowflake\Pool;
use Exception;
use JetBrains\PhpStorm\Pure;
use Snowflake\Abstracts\Component;
use Snowflake\Abstracts\Config;
use Snowflake\Exception\ConfigException;
use Snowflake\Snowflake;
use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use Swoole\Timer;
/**
* Class ClientsPool
* @package Snowflake\Pool
*/
class ClientsPool extends Component
{
/** @var Channel[] */
private static array $_connections = [];
public int $max = 60;
public int $creates = -1;
private array $_times = [];
protected static array $hasCreate = [];
/**
* @param string $name
*/
public function increment(string $name)
{
if (!isset(static::$hasCreate[$name])) {
static::$hasCreate[$name] = 0;
}
static::$hasCreate[$name] += 1;
}
/**
* @param string $name
*/
public function decrement(string $name)
{
if (!isset(static::$hasCreate[$name])) {
return;
}
if (static::$hasCreate[$name] <= 0) {
return;
}
static::$hasCreate[$name] -= 1;
}
/**
* @return array
* @throws ConfigException
*/
private function getClearTime(): array
{
$firstClear = Config::get('pool.clear.start', 600);
$lastClear = Config::get('pool.clear.end', 300);
return [$firstClear, $lastClear];
}
/**
* @throws Exception
*/
public function Heartbeat_detection($ticker)
{
if (env('state') == 'exit') {
Timer::clear($this->creates);
foreach (static::$_connections as $channel) {
$this->flush($channel, 0);
$channel->close();
}
static::$_connections = [];
$this->creates = -1;
} else {
$this->heartbeat_flush();
}
}
/**
* @throws ConfigException
* @throws Exception
*/
private function heartbeat_flush()
{
$num = [];
$total = 0;
$min = Config::get('databases.pool.min', 1);
foreach (static::$_connections as $key => $channel) {
if (!isset($num[$key])) {
$num[$key] = 0;
}
if (time() - ($this->_times[$key] ?? time()) > 120) {
$this->flush($channel, 0);
} else if ($channel->length() > $min) {
$this->flush($channel, $min);
}
$num[$key] += ($length = $channel->length());
if (str_starts_with($key, 'Mysql') && (Snowflake::isWorker() || Snowflake::isTask()) && $length > 0) {
$this->debug('Worker #' . env('worker') . ' use client -> ' . $key . ':' . $length);
}
$total += $length;
}
write(var_export($num, true), 'connections');
if ($total < 1) {
Timer::clear($this->creates);
if (Snowflake::isWorker() || Snowflake::isTask()) {
$this->debug('Worker #' . env('worker') . ' clear time tick.');
}
$this->creates = -1;
}
}
/**
* @param $channel
* @param $retain_number
* @throws Exception
*/
public function flush($channel, $retain_number)
{
$this->pop($channel, $retain_number);
}
/**
* @param Channel $channel
* @param $retain_number
* @throws Exception
*/
protected function pop(Channel $channel, $retain_number): void
{
if (Coroutine::getCid() === -1) {
return;
}
while ($channel->length() > $retain_number) {
$connection = $channel->pop();
if ($connection) {
unset($connection);
}
}
}
/**
* @param $name
* @param false $isMaster
* @param int $max
*/
public function initConnections($name, bool $isMaster = false, int $max = 60)
{
$name = $this->name($name, $isMaster);
if (isset(static::$_connections[$name]) && static::$_connections[$name] instanceof Channel) {
return;
}
if (Coroutine::getCid() === -1) {
return;
}
if ($this->creates === -1) {
$this->creates = Timer::tick(60000, [$this, 'Heartbeat_detection']);
}
static::$_connections[$name] = new Channel($max);
$this->max = $max;
}
/**
* @param $name
* @return Channel
* @throws ConfigException
* @throws Exception
*/
private function getChannel($name): Channel
{
if (!isset(static::$_connections[$name])) {
static::$_connections[$name] = new Channel(Config::get('databases.pool.max', 10));
}
if (static::$_connections[$name]->errCode == SWOOLE_CHANNEL_CLOSED){
throw new Exception('Channel is Close.');
}
if ($this->creates === -1) {
$this->creates = Timer::tick(60000, [$this, 'Heartbeat_detection']);
}
return static::$_connections[$name];
}
/**
* @param $name
* @return array
* @throws Exception
*/
public function getFromChannel($name): mixed
{
$this->_times[$name] = time();
$channel = $this->getChannel($name);
if (!$channel->isEmpty()) {
$connection = $channel->pop();
if ($this->checkCanUse($name, $connection)) {
return $connection;
}
}
return null;
}
/**
* @param $cds
* @param false $isMaster
* @return string
*/
#[Pure] public function name($cds, bool $isMaster = false): string
{
if ($isMaster === true) {
return $cds . '_master';
} else {
return $cds . '_slave';
}
}
/**
* @param string $name
* @param mixed $client
* @return bool
* 检查连接可靠性
*/
public function checkCanUse(string $name, mixed $client): bool
{
return true;
}
/**
* @param array $config
* @param bool $isMaster
* @return mixed
* @throws Exception
*/
public function get(mixed $config, bool $isMaster): mixed
{
throw new Exception('Undefined system processing function.');
}
/**
* @param string $name
* @return bool
*/
public function hasItem(string $name): bool
{
if (isset(static::$_connections[$name])) {
return !static::$_connections[$name]->isEmpty();
}
return false;
}
/**
* @param string $name
* @return mixed
*/
public function size(string $name): mixed
{
if (Coroutine::getCid() === -1) {
return 0;
}
if (!isset(static::$_connections[$name])) {
return 0;
}
return static::$_connections[$name]->length();
}
/**
* @param string $name
* @param mixed $client
* @throws ConfigException
*/
public function push(string $name, mixed $client)
{
if (Coroutine::getCid() === -1) {
return;
}
$channel = $this->getChannel($name);
if (!$channel->isFull()) {
$channel->push($client);
}
unset($client);
}
/**
* @param string $name
* @throws Exception
*/
public function clean(string $name)
{
if (Coroutine::getCid() === -1 || !isset(static::$_connections[$name])) {
return;
}
$channel = static::$_connections[$name];
$this->pop($channel, 0);
}
/**
* @return Channel[]
*/
protected function getChannels(): array
{
return static::$_connections;
}
}
System/Pool/Connection.php
+3 -3
View File
@@ -20,7 +20,7 @@ use Throwable;
class Connection extends Component
{
private ?ClientsPool $clientsPool = null;
private ?Pool $clientsPool = null;
/**
@@ -247,10 +247,10 @@ class Connection extends Component
/**
* @return ClientsPool
* @return Pool
* @throws Exception
*/
public function getPool(): ClientsPool
public function getPool(): Pool
{
if (!$this->clientsPool) {
$this->clientsPool = Snowflake::app()->getClientsPool();
System/Pool/Pool.php
+309 -17
View File
@@ -1,40 +1,332 @@
<?php
declare(strict_types=1);
namespace Snowflake\Pool;
use Exception;
use Snowflake\Abstracts\BaseObject;
use JetBrains\PhpStorm\Pure;
use Snowflake\Abstracts\Component;
use Snowflake\Abstracts\Config;
use Snowflake\Exception\ConfigException;
use Snowflake\Snowflake;
use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use Swoole\Timer;
/**
* Class Pool
* @package Snowflake\Pool
* @property Redis $redis
* @property Connection $db
* @property $memcached
*/
class Pool extends BaseObject
class Pool extends Component
{
/**
* @return Redis
* @throws Exception
*/
public function getRedis(): Redis
{
return Snowflake::app()->get('redis_connections');
}
/** @var Channel[] */
private static array $_connections = [];
public int $max = 60;
public int $creates = -1;
private array $_times = [];
protected static array $hasCreate = [];
/**
* @param string $name
*/
public function increment(string $name)
{
if (!isset(static::$hasCreate[$name])) {
static::$hasCreate[$name] = 0;
}
static::$hasCreate[$name] += 1;
}
/**
* @param string $name
*/
public function decrement(string $name)
{
if (!isset(static::$hasCreate[$name])) {
return;
}
if (static::$hasCreate[$name] <= 0) {
return;
}
static::$hasCreate[$name] -= 1;
}
/**
* @return array
* @throws ConfigException
*/
private function getClearTime(): array
{
$firstClear = Config::get('pool.clear.start', 600);
$lastClear = Config::get('pool.clear.end', 300);
return [$firstClear, $lastClear];
}
/**
* @return Connection
* @throws Exception
*/
public function getDb(): Connection
public function Heartbeat_detection($ticker)
{
return Snowflake::app()->get('connections');
if (env('state') == 'exit') {
Timer::clear($this->creates);
foreach (static::$_connections as $channel) {
$this->flush($channel, 0);
$channel->close();
}
static::$_connections = [];
$this->creates = -1;
} else {
$this->heartbeat_flush();
}
}
/**
* @throws ConfigException
* @throws Exception
*/
private function heartbeat_flush()
{
$num = [];
$total = 0;
$min = Config::get('databases.pool.min', 1);
foreach (static::$_connections as $key => $channel) {
if (!isset($num[$key])) {
$num[$key] = 0;
}
if (time() - ($this->_times[$key] ?? time()) > 120) {
$this->flush($channel, 0);
} else if ($channel->length() > $min) {
$this->flush($channel, $min);
}
$num[$key] += ($length = $channel->length());
if (str_starts_with($key, 'Mysql') && (Snowflake::isWorker() || Snowflake::isTask()) && $length > 0) {
$this->debug('Worker #' . env('worker') . ' use client -> ' . $key . ':' . $length);
}
$total += $length;
}
write(var_export($num, true), 'connections');
if ($total < 1) {
Timer::clear($this->creates);
if (Snowflake::isWorker() || Snowflake::isTask()) {
$this->debug('Worker #' . env('worker') . ' clear time tick.');
}
$this->creates = -1;
}
}
/**
* @param $channel
* @param $retain_number
* @throws Exception
*/
public function flush($channel, $retain_number)
{
$this->pop($channel, $retain_number);
}
/**
* @param Channel $channel
* @param $retain_number
* @throws Exception
*/
protected function pop(Channel $channel, $retain_number): void
{
if (Coroutine::getCid() === -1) {
return;
}
while ($channel->length() > $retain_number) {
$connection = $channel->pop();
if ($connection) {
unset($connection);
}
}
}
/**
* @param $name
* @param false $isMaster
* @param int $max
*/
public function initConnections($name, bool $isMaster = false, int $max = 60)
{
$name = $this->name($name, $isMaster);
if (isset(static::$_connections[$name]) && static::$_connections[$name] instanceof Channel) {
return;
}
if (Coroutine::getCid() === -1) {
return;
}
if ($this->creates === -1) {
$this->creates = Timer::tick(60000, [$this, 'Heartbeat_detection']);
}
static::$_connections[$name] = new Channel($max);
$this->max = $max;
}
/**
* @param $name
* @return Channel
* @throws ConfigException
* @throws Exception
*/
private function getChannel($name): Channel
{
if (!isset(static::$_connections[$name])) {
static::$_connections[$name] = new Channel(Config::get('databases.pool.max', 10));
}
if (static::$_connections[$name]->errCode == SWOOLE_CHANNEL_CLOSED){
throw new Exception('Channel is Close.');
}
if ($this->creates === -1) {
$this->creates = Timer::tick(60000, [$this, 'Heartbeat_detection']);
}
return static::$_connections[$name];
}
/**
* @param $name
* @return array
* @throws Exception
*/
public function getFromChannel($name): mixed
{
$this->_times[$name] = time();
$channel = $this->getChannel($name);
if (!$channel->isEmpty()) {
$connection = $channel->pop();
if ($this->checkCanUse($name, $connection)) {
return $connection;
}
}
return null;
}
/**
* @param $cds
* @param false $isMaster
* @return string
*/
#[Pure] public function name($cds, bool $isMaster = false): string
{
if ($isMaster === true) {
return $cds . '_master';
} else {
return $cds . '_slave';
}
}
/**
* @param string $name
* @param mixed $client
* @return bool
* 检查连接可靠性
*/
public function checkCanUse(string $name, mixed $client): bool
{
return true;
}
/**
* @param array $config
* @param bool $isMaster
* @return mixed
* @throws Exception
*/
public function get(mixed $config, bool $isMaster): mixed
{
throw new Exception('Undefined system processing function.');
}
/**
* @param string $name
* @return bool
*/
public function hasItem(string $name): bool
{
if (isset(static::$_connections[$name])) {
return !static::$_connections[$name]->isEmpty();
}
return false;
}
/**
* @param string $name
* @return mixed
*/
public function size(string $name): mixed
{
if (Coroutine::getCid() === -1) {
return 0;
}
if (!isset(static::$_connections[$name])) {
return 0;
}
return static::$_connections[$name]->length();
}
/**
* @param string $name
* @param mixed $client
* @throws ConfigException
*/
public function push(string $name, mixed $client)
{
if (Coroutine::getCid() === -1) {
return;
}
$channel = $this->getChannel($name);
if (!$channel->isFull()) {
$channel->push($client);
}
unset($client);
}
/**
* @param string $name
* @throws Exception
*/
public function clean(string $name)
{
if (Coroutine::getCid() === -1 || !isset(static::$_connections[$name])) {
return;
}
$channel = static::$_connections[$name];
$this->pop($channel, 0);
}
/**
* @return Channel[]
*/
protected function getChannels(): array
{
return static::$_connections;
}
}
System/Pool/Redis.php
+3 -3
View File
@@ -22,7 +22,7 @@ use Swoole\Runtime;
class Redis extends Component
{
private ?ClientsPool $clientsPool = null;
private ?Pool $clientsPool = null;
/**
@@ -114,10 +114,10 @@ class Redis extends Component
/**
* @return ClientsPool
* @return Pool
* @throws Exception
*/
public function getPool(): ClientsPool
public function getPool(): Pool
{
if (!$this->clientsPool) {
$this->clientsPool = Snowflake::app()->getClientsPool();