Merge pull request #155 from xudianyang/master

优化连接池，提供max_conn,max_time,min_conn选项

Author: xudianyang

src/Helpers/Common.php

@@ -23,6 +23,11 @@ abstract class AsynPool implements IAsynPool
      */
     const MAX_TOKEN = 655360;
 
+    /**
+     * 连接池类型名称
+     */
+    const ASYN_NAME = '';
+
     /**
      * @var Config 配置对象
      */
@@ -68,22 +73,34 @@ abstract class AsynPool implements IAsynPool
      */
     protected $waitConnectNum = 0;
 
+    /**
+     * @var int 连接峰值
+     */
+    protected $establishedConn = 0;
+
     /**
      * @var AsynPoolManager 连接池管理器
      */
     protected $asynManager;
 
+    /**
+     * @var string 连接池标识
+     */
+    protected $active;
+
     /**
      * AsynPool constructor.
      *
      * @param Config $config 配置对象
+     * @param string $active 连接池名称
      */
-    public function __construct($config)
+    public function __construct($config, $active)
     {
         $this->callBacks = [];
         $this->commands  = new \SplQueue();
         $this->pool      = new \SplQueue();
         $this->config    = $config;
+        $this->active    = $active;
     }
 
     /**
@@ -158,14 +175,58 @@ public function workerInit($workerId)
      */
     public function pushToPool($client)
     {
-        $this->pool->push($client);
-        if (count($this->commands) > 0) {//有残留的任务
-            $command = $this->commands->shift();
-            $this->execute($command);
+        $maxTime = $this->config[static::ASYN_NAME][$this->active]['max_time'] ?? 3600;
+        $minConn = $this->config[static::ASYN_NAME][$this->active]['min_conn'] ?? 0;
+
+        //回归连接
+        if (((time() - $client->genTime) < $maxTime)
+            || (($this->establishedConn + $this->waitConnectNum) <= $minConn)
+        ) {
+            $this->pool->push($client);
+            if (count($this->commands) > 0) {//有残留的任务
+                $command = $this->commands->shift();
+                $this->execute($command);
+            }
+        } else {
+            $client->close();
         }
+
         return $this;
     }
 
+    /**
+     * 创建一个连接
+     */
+    public function prepareOne()
+    {
+        $maxConn = $this->config[static::ASYN_NAME][$this->active]['max_conn'] ?? null;
+        if ($maxConn) {
+            if ($maxConn > ($this->waitConnectNum + $this->establishedConn)) {
+                $this->reconnect();
+            }
+        } else {
+            $this->reconnect();
+        }
+    }
+
+    /**
+     * 返回唯一的连接池名称
+     *
+     * @return string
+     */
+    public function getAsynName()
+    {
+        return self::ASYN_NAME  . '.' . $this->active;
+    }
+
+    /**
+     * 建立连接
+     *
+     * @param null $client
+     * @return mixed
+     */
+    abstract public function reconnect($client = null);
+
     /**
      * 获取同步
      *

src/Pools/AsynPool.php

@@ -84,4 +84,12 @@ public function addTokenCallback($callback);
      * @throws Exception
      */
     public function getSync();
+
+    /**
+     * 建立连接
+     *
+     * @param null $client
+     * @return mixed
+     */
+    public function reconnect($client = null);
 }

src/Pools/IAsynPool.php

@@ -26,7 +26,7 @@ class MysqlAsynPool extends AsynPool
     /**
      * 连接池类型名称
      */
-    const ASYN_NAME = 'mysql.';
+    const ASYN_NAME = 'mysql';
 
     /**
      * @var Miner SQL Builder
@@ -38,16 +38,6 @@ class MysqlAsynPool extends AsynPool
      */
     public $bindPool;
 
-    /**
-     * @var int 连接峰值
-     */
-    protected $mysqlMaxCount = 0;
-
-    /**
-     * @var string 连接池标识
-     */
-    private $active;
-
     /**
      * MysqlAsynPool constructor.
      *
@@ -56,8 +46,7 @@ class MysqlAsynPool extends AsynPool
      */
     public function __construct($config, $active)
     {
-        parent::__construct($config);
-        $this->active         = $active;
+        parent::__construct($config, $active);
         $this->bindPool       = [];
     }
 
@@ -161,15 +150,7 @@ public function execute($data)
 
             //不是绑定的连接就回归连接
             if (!isset($data['bind_id'])) {
-                //回归连接
-                if (((time() - $client->genTime) < 3600)
-                    || (($this->mysqlMaxCount + $this->waitConnectNum) <= 30)
-                ) {
-                    $this->pushToPool($client);
-                } else {
-                    $client->close();
-                    $this->mysqlMaxCount--;
-                }
+                $this->pushToPool($client);
             } else {//事务
                 $bindId = $data['bind_id'];
                 if ($sql == 'commit' || $sql == 'rollback') {//结束事务
@@ -179,14 +160,6 @@ public function execute($data)
         });
     }
 
-    /**
-     * 创建一个Mysql连接
-     */
-    public function prepareOne()
-    {
-        $this->reconnect();
-    }
-
     /**
      * 重连或者连接
      *
@@ -207,8 +180,8 @@ public function reconnect($client = null)
             } else {
                 $client->isClose = false;
                 if (!isset($client->client_id)) {
-                    $client->client_id = $this->mysqlMaxCount;
-                    $this->mysqlMaxCount++;
+                    $client->client_id = $this->establishedConn;
+                    $this->establishedConn++;
                 }
                 $this->pushToPool($client);
             }
@@ -239,19 +212,10 @@ public function freeBind($bindId, Context $context = null)
      */
     public function onClose($client)
     {
+        $this->establishedConn--;
         $client->isClose = true;
     }
 
-    /**
-     * 返回唯一的连接池名称
-     *
-     * @return string
-     */
-    public function getAsynName()
-    {
-        return self::ASYN_NAME . $this->active;
-    }
-
     /**
      * 开启一个同步事务
      *

src/Pools/MysqlAsynPool.php

@@ -24,23 +24,13 @@ class RedisAsynPool extends AsynPool
     /**
      * 连接池类型名称
      */
-    const ASYN_NAME = 'redis.';
+    const ASYN_NAME = 'redis';
 
     /**
      * @var array 连接配置信息
      */
     public $connect;
 
-    /**
-     * @var int 连接峰值
-     */
-    protected $redisMaxCount = 0;
-
-    /**
-     * @var string 连接池标识
-     */
-    private $active;
-
     /**
      * @var CoroutineRedisProxy 连接池辅助类
      */
@@ -80,8 +70,7 @@ class RedisAsynPool extends AsynPool
      */
     public function __construct($config, string $active)
     {
-        parent::__construct($config);
-        $this->active = $active;
+        parent::__construct($config, $active);
 
         $config = $this->config->get('redis.' . $this->active, null);
         if (!$config) {
@@ -456,27 +445,12 @@ public function execute($data)
                 //给worker发消息
                 $this->asynManager->sendMessageToWorker($this, $data);
                 //回归连接
-                if (((time() - $client->genTime) < 3600)
-                    || (($this->redisMaxCount + $this->waitConnectNum) <= 30)
-                ) {
-                    $this->pushToPool($client);
-                } else {
-                    $client->close();
-                    $this->redisMaxCount--;
-                }
+                $this->pushToPool($client);
             };
             $client->__call($data['name'], array_values($arguments));
         }
     }
 
-    /**
-     * 创建一个Redis连接
-     */
-    public function prepareOne()
-    {
-        $this->reconnect();
-    }
-
     /**
      * 重连或者连接
      *
@@ -512,8 +486,8 @@ public function reconnect($client = null)
 
             $client->isClose = false;
             if (!isset($client->client_id)) {
-                $client->client_id = $this->redisMaxCount;
-                $this->redisMaxCount++;
+                $client->client_id = $this->establishedConn;
+                $this->establishedConn++;
             }
             $this->pushToPool($client);
         });
@@ -527,15 +501,6 @@ public function reconnect($client = null)
     public function onClose($client)
     {
         $client->isClose = true;
-    }
-
-    /**
-     * 返回唯一的连接池名称
-     *
-     * @return string
-     */
-    public function getAsynName()
-    {
-        return self::ASYN_NAME . $this->active;
+        $this->establishedConn--;
     }
 }