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-08-16 18:43:37 +08:00
parent cb9e7e8775
commit 1a89c6c8be
3 changed files with 216 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
System/Cache/Base/Redis.php
+145
View File
@@ -0,0 +1,145 @@
<?php
namespace Kiri\Cache\Base;
use HttpServer\Http\Context;
use Kiri\Events\EventProvider;
use Kiri\Exception\RedisConnectException;
use Kiri\Kiri;
use Server\Events\OnWorkerExit;
use Swoole\Timer;
class Redis
{
const DB_ERROR_MESSAGE = 'The system is busy, please try again later.';
private ?\Redis $pdo = null;
private int $_transaction = 0;
/**
* @var EventProvider
*/
private EventProvider $eventProvider;
private int $_timer = -1;
private int $_last = 0;
/**
* @param string $host
* @param int $port
* @param int $database
* @param string $auth
* @param string $prefix
* @param int $timeout
* @param int $read_timeout
* @throws \ReflectionException
*/
public function __construct(public string $host, public int $port, public int $database = 0,
public string $auth = '', public string $prefix = '', public int $timeout = 30,
public int $read_timeout = 30)
{
$this->eventProvider = Kiri::getDi()->get(EventProvider::class);
}
public function init()
{
$this->heartbeat_check();
$this->eventProvider->on(OnWorkerExit::class, [$this, 'stopHeartbeatCheck']);
}
/**
*
*/
public function heartbeat_check(): void
{
if ($this->_timer === -1 && Context::inCoroutine()) {
$this->_timer = Timer::tick(3000, function () {
try {
if (time() - $this->_last > 10 * 60) {
$this->stopHeartbeatCheck();
$this->pdo = null;
}
} catch (\Throwable $throwable) {
error($throwable);
}
});
}
}
/**
*
*/
public function stopHeartbeatCheck(): void
{
if (Context::inCoroutine()) {
Timer::clear($this->_timer);
}
$this->_timer = -1;
}
/**
* @param string $name
* @param array $arguments
* @return mixed
* @throws RedisConnectException
*/
public function __call(string $name, array $arguments)
{
if (!method_exists($this, $name)) {
return $this->_pdo()->{$name}(...$arguments);
}
return $this->{$name}(...$arguments);
}
/**
* @return \Redis
* @throws RedisConnectException
*/
public function _pdo(): \Redis
{
if ($this->_timer === -1) {
$this->heartbeat_check();
}
if (!($this->pdo instanceof \Redis)) {
$this->pdo = $this->newClient();
}
return $this->pdo;
}
/**
* @return \Redis
* @throws RedisConnectException
*/
private function newClient(): \Redis
{
$redis = new \Redis();
if (!$redis->pconnect($this->host, $this->port, $this->timeout)) {
throw new RedisConnectException(sprintf('The Redis Connect %s::%d Fail.', $this->host, $this->port));
}
if (!empty($config['auth']) && !$redis->auth($config['auth'])) {
throw new RedisConnectException(sprintf('Redis Error: %s, Host %s, Auth %s', $redis->getLastError(), $this->host, $this->auth));
}
if ($this->read_timeout < 0) {
$this->read_timeout = 0;
}
$redis->select($this->database);
$redis->setOption(\Redis::OPT_READ_TIMEOUT, $this->read_timeout);
$redis->setOption(\Redis::OPT_PREFIX, $this->prefix);
return $redis;
}
}
System/Pool/Pool.php
-3
View File
@@ -27,8 +27,6 @@ class Pool extends Component
public int $max = 60;
private array $_times = [];
use Alias;
@@ -110,7 +108,6 @@ class Pool extends Component
if (Coroutine::getCid() === -1) {
return $callback();
}
$this->_times[$name] = time();
$channel = $this->getChannel($name);
if (!$channel->isEmpty()) {
$connection = $channel->pop();
System/Pool/Redis.php
+71 -86
View File
@@ -8,11 +8,11 @@ namespace Kiri\Pool;
use Closure;
use Exception;
use HttpServer\Http\Context;
use Redis as SRedis;
use Kiri\Abstracts\Component;
use Kiri\Exception\ConfigException;
use Kiri\Exception\RedisConnectException;
use Kiri\Kiri;
use Redis as SRedis;
use Swoole\Coroutine;
use Swoole\Runtime;
@@ -23,24 +23,24 @@ use Swoole\Runtime;
class Redis extends Component
{
use Alias;
use Alias;
/**
* @param mixed $config
* @param bool $isMaster
* @return mixed
* @throws Exception
*/
public function get(mixed $config, bool $isMaster = false): mixed
{
$coroutineName = $this->name('Redis:' . $config['host'], $isMaster);
if (Context::hasContext($coroutineName)) {
return Context::getContext($coroutineName);
}
$clients = $this->getPool()->get($coroutineName, $this->create($coroutineName, $config));
return Context::setContext($coroutineName, $clients);
}
/**
* @param mixed $config
* @param bool $isMaster
* @return mixed
* @throws Exception
*/
public function get(mixed $config, bool $isMaster = false): mixed
{
$coroutineName = $this->name('Redis:' . $config['host'], $isMaster);
if (Context::hasContext($coroutineName)) {
return Context::getContext($coroutineName);
}
$clients = $this->getPool()->get($coroutineName, $this->create($coroutineName, $config));
return Context::setContext($coroutineName, $clients);
}
/**
@@ -48,83 +48,68 @@ class Redis extends Component
* @param mixed $config
* @return Closure
*/
public function create(string $name, mixed $config): Closure
{
return static function () use ($name, $config) {
if (Coroutine::getCid() === -1) {
Runtime::enableCoroutine(false);
}
$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']);
return $redis;
};
}
public function create(string $name, mixed $config): Closure
{
return static function () use ($name, $config) {
return Kiri::getDi()->newObject(\Kiri\Cache\Base\Redis::class, [
$config['host'], (int)$config['port'], $config['databases'] ?? 0,
$config['auth'], $config['prefix'] ?? '', $config['timeout'] ?? 30,
$config['read_timeout'] ?? 30
]);
};
}
/**
* @param array $config
* @param bool $isMaster
* @throws ConfigException
* @throws Exception
*/
public function release(array $config, bool $isMaster = false)
{
$coroutineName = $this->name('Redis:' . $config['host'], $isMaster);
if (!Context::hasContext($coroutineName)) {
return;
}
/**
* @param array $config
* @param bool $isMaster
* @throws ConfigException
* @throws Exception
*/
public function release(array $config, bool $isMaster = false)
{
$coroutineName = $this->name('Redis:' . $config['host'], $isMaster);
if (!Context::hasContext($coroutineName)) {
return;
}
$this->getPool()->push($coroutineName, Context::getContext($coroutineName));
Context::remove($coroutineName);
}
$this->getPool()->push($coroutineName, Context::getContext($coroutineName));
Context::remove($coroutineName);
}
/**
* @param array $config
* @param bool $isMaster
* @throws Exception
*/
public function destroy(array $config, bool $isMaster = false)
{
$coroutineName = $this->name('Redis:' . $config['host'], $isMaster);
if (Context::hasContext($coroutineName)) {
$this->getPool()->decrement($coroutineName);
}
$this->getPool()->clean($coroutineName);
Context::remove($coroutineName);
}
/**
* @param array $config
* @param bool $isMaster
* @throws Exception
*/
public function destroy(array $config, bool $isMaster = false)
{
$coroutineName = $this->name('Redis:' . $config['host'], $isMaster);
$this->getPool()->clean($coroutineName);
Context::remove($coroutineName);
}
/**
* @return Pool
* @throws Exception
*/
public function getPool(): Pool
{
return Kiri::getDi()->get(Pool::class);
}
/**
* @return Pool
* @throws Exception
*/
public function getPool(): Pool
{
return Kiri::getDi()->get(Pool::class);
}
/**
* @param $name
* @param $isMaster
* @param $max
* @throws Exception
*/
public function initConnections($name, $isMaster, $max)
{
$this->getPool()->initConnections($name, $isMaster, $max);
}
/**
* @param $name
* @param $isMaster
* @param $max
* @throws Exception
*/
public function initConnections($name, $isMaster, $max)
{
$this->getPool()->initConnections($name, $isMaster, $max);
}
}