From 1a89c6c8be55d3013ec50983344924dff0d2d22f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mr=C2=B7x?= Date: Mon, 16 Aug 2021 18:43:37 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E5=90=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- System/Cache/Base/Redis.php | 145 +++++++++++++++++++++++++++++++++ System/Pool/Pool.php | 3 - System/Pool/Redis.php | 157 ++++++++++++++++-------------------- 3 files changed, 216 insertions(+), 89 deletions(-) create mode 100644 System/Cache/Base/Redis.php diff --git a/System/Cache/Base/Redis.php b/System/Cache/Base/Redis.php new file mode 100644 index 00000000..41f36b40 --- /dev/null +++ b/System/Cache/Base/Redis.php @@ -0,0 +1,145 @@ +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; + + } + +} diff --git a/System/Pool/Pool.php b/System/Pool/Pool.php index ab5be537..8bf82525 100644 --- a/System/Pool/Pool.php +++ b/System/Pool/Pool.php @@ -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(); diff --git a/System/Pool/Redis.php b/System/Pool/Redis.php index 0deddf52..61f52b85 100644 --- a/System/Pool/Redis.php +++ b/System/Pool/Redis.php @@ -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); + } }