用Swoole4 打造高并发的PHP协程Mysql连接池

Posted 2021-02-28 xiami2046

码云代码仓库：https://gitee.com/tanjiajun/MysqlPool

代码仓库：https://github.com/asbectJ/swoole4.git

前言

在写这篇文章之前，看了好几篇实现连接池的文章，都是写的很不好的。摆明忽略了连接池的很多特性，很多都不具有抗高并发和连接复用。所以自己觉得有必须把最近几天，实现一个比较完整的php数据库连接池的点滴记录下来，望能帮助各位，感激者望多点赞和打赏。

一、数据库连接池基本概念

所谓的数据库连接池，一般指的就是程序和数据库保持一定数量的数据库连接不断开，并且各请求的连接可以相互复用，减少重复新建数据库连接的消耗和避免在高并发的情况下出现数据库max connections等错误。自己总结了一下，如果要实现一个数据库连接池，一般有几个特点：

• 连接复用，不同的请求连接，可以放回池中，等待下个请求发分配和调用
• 连接数量一般维持min-max的最大最少值之间
• 对于空闲连接的回收
• 可以抗一定程度的高并发，也就是说当一次并发请求完池中所有的连接时，获取不到连接的请求可等待其他连接的释放

总结几个特性后，一个基本连接池，大致要实现下图功能：

 

1. 创建连接：连接池启动后，初始化一定的空闲连接，指定为最少的连接min。当连接池为空，不够用时，创建新的连接放到池里，但不能超过指定的最大连接max数量。
2. 连接释放：每次使用完连接，一定要调用释放方法，把连接放回池中，给其他程序或请求使用。
3. 连接分配：连接池中用pop和push的方式对等入队和出队分配与回收。能实现阻塞分配，也就是在池空并且已创建数量大于max，阻塞一定时间等待其他请求的连接释放，超时则返回null。
4. 连接管理：对连接池中的连接，定时检活和释放空闲连接等

二、Fpm+数据库长连接的实现

1. 利用fpm实现：例如你要实例一个100连接数的池，开启100个空闲fpm,然后每个fpm的连接都是数据库长连接。一般pm.max_spare_servers = 8这个配置项就是维持连接池的空闲数量，然后pm.max_children = 50就是最大的连接数量。和fpm的进程数量一致。

三、基于swoole的实现

• swoole简单介绍（更多参阅swoole官网）

      swoole是一个PHP实现异步网络通信的引擎或者扩展，其中实现了很多传统PHP-fpm没有的东西，例如异步的客户端，异步Io，常驻内存，协程等等，一个个优秀的扩展，其中异步和协程等概念能应用于高并发场景。缺点是文档和入门的门槛都比较高，需要排坑。附上swoole的运行流程和进程结构图：

运行流程图

进程/线程架构图

• 基于swoole现实时的注意事项

首先，为了减少大家对之后运行示例代码产生不必要的天坑，先把注意事项和场景问题放前面：

1、程序中使用了协程的通信管道channel（与go的chan差不多的），其中swoole2是不支持chan->pop($timeout)中timeout超时等待的，所以必须用swoole4版本

2、使用swoole协程扩展的时候，一定不能装xdebug之类的扩展，否则报错。官方说明为：https://wiki.swoole.com/wiki/page/674.html，同时参考如下了解更多关于swoole协程的使用和注意：https://wiki.swoole.com/wiki/page/749.html

3、笔者使用的环境为：PHP 7.1.18和swoole4作为此次开发的环境

• 基于swoole现实连接池的方法

首先，此次利用swoole实现连接池，运用到swoole以下技术或者概念

1、连接变量池，这里可以看做一个数组或者队列，利用swoole全局变量的常驻内存特性，只要变量没主动unset掉，数组或队列中的连接对象可以一直保持，不释放。主要参考：https://wiki.swoole.com/wiki/page/p-zend_mm.html

2、协程。协程是纯用户状态的线程，通过协作的方式而不是抢占的方式来切换。首先此次的连接池两处用到协程：

• 一个是mysql的协程客户端，为什么要用协程客户端，因为如果是用同步客户端PDO，在一个进程处理内，就算有几百个连接池，swoole worker进程中用普通的PDO方式,随便并发多少个请求，每一个请求都只能等上一个请求执行完毕，woker才处理下一个请求，这里就算阻塞了。为了让一个worker支持阻塞切换出cpu去处理其他请求，所以要用到协程的协助切换，或者异步客户端也可以，但是异步客户端使用起来嵌套太多，很不方便。swoole协程可以无感知的用同步的代码编写方式达到异步IO的效果和性能。
• 第二个是底层实现了协程切换和调度的channel，以下详述什么是channel

3、Coroutine/channel通道，类似于go语言的chan，支持多生产者协程和多消费者协程。底层自动实现了协程的切换和调度。高并发时，容易出连接池为空时，如果用一般的array或者splqueue()作为介质存储连接对象变量，不能产生阻塞等待其他请求释放的效果，也就是说只能直接返回null.。所以这里用了一个swoole4协程中很牛逼的channel通过管道作为存储介质，它的出队方法pop($timeout)可以指定阻塞等待指定时间后返回。注意，是swoole2是没有超时timeout的参数，不适用此场景。在go语言中，如果chan等待或者push了没有消费或者生产一对一的情况，是会发生死锁。所以swoole4的timeout应该是为了避免无限等待为空channel情况而产生。主要参考：

https://wiki.swoole.com/wiki/page/p-coroutine_channel.html

channel切换的例子：

1.  
   <?php
2.  
   use SwooleCoroutineChannel;
3.  
   $chan = new Channel();
4.  
   go(function () use ($chan) {
5.  
   echo "我是第一个协程，等待3秒内有push就执行返回" . PHP_EOL;
6.  
   $p = $chan->pop(2);#1
7.  
   echo "pop返回结果" . PHP_EOL;
8.  
   var_dump($p);
9.  
   });
10.  
    go(function () use ($chan) {
11.  
    co::sleep(1);#2
12.  
    $chan->push(1);
13.  
    });
14.  
    echo "main" . PHP_EOL;

#1处代码会首先执行，然后遇到pop(),因为channel还是空，会等待2s。此时协程会让出cpu,跳到第二个协程执行，然后#2出睡眠1秒，push变量1进去channel后返回#1处继续执行，成功取车通过中刚push的值1.运行结果为：

如果把#2处的睡眠时间换成大于pop()的等待时间，结果是：

• 根据这些特性最终实现连接池的抽象封装类为：

1.  
   <?php
2.  
   /**
3.  
   * 连接池封装.
4.  
   * User: user
5.  
   * Date: 2018/9/1
6.  
   * Time: 13:36
7.  
   */
8.  
    
9.  
   use SwooleCoroutineChannel;
10.  
     
11.  
    abstract class AbstractPool
12.  
    {
13.  
    private $min;//最少连接数
14.  
    private $max;//最大连接数
15.  
    private $count;//当前连接数
16.  
    private $connections;//连接池组
17.  
    protected $spareTime;//用于空闲连接回收判断
18.  
    //数据库配置
19.  
    protected $dbConfig = array(
20.  
    ‘host‘ => ‘10.0.2.2‘,
21.  
    ‘port‘ => 3306,
22.  
    ‘user‘ => ‘root‘,
23.  
    ‘password‘ => ‘root‘,
24.  
    ‘database‘ => ‘test‘,
25.  
    ‘charset‘ => ‘utf8‘,
26.  
    ‘timeout‘ => 2,
27.  
    );
28.  
     
29.  
    private $inited = false;
30.  
     
31.  
    protected abstract function createDb();
32.  
     
33.  
    public function __construct()
34.  
    {
35.  
    $this->min = 10;
36.  
    $this->max = 100;
37.  
    $this->spareTime = 10 * 3600;
38.  
    $this->connections = new Channel($this->max + 1);
39.  
    }
40.  
     
41.  
    protected function createObject()
42.  
    {
43.  
    $obj = null;
44.  
    $db = $this->createDb();
45.  
    if ($db) {
46.  
    $obj = [
47.  
    ‘last_used_time‘ => time(),
48.  
    ‘db‘ => $db,
49.  
    ];
50.  
    }
51.  
    return $obj;
52.  
    }
53.  
     
54.  
    /**
55.  
    * 初始换最小数量连接池
56.  
    * @return $this|null
57.  
    */
58.  
    public function init()
59.  
    {
60.  
    if ($this->inited) {
61.  
    return null;
62.  
    }
63.  
    for ($i = 0; $i < $this->min; $i++) {
64.  
    $obj = $this->createObject();
65.  
    $this->count++;
66.  
    $this->connections->push($obj);
67.  
    }
68.  
    return $this;
69.  
    }
70.  
     
71.  
    public function getConnection($timeOut = 3)
72.  
    {
73.  
    $obj = null;
74.  
    if ($this->connections->isEmpty()) {
75.  
    if ($this->count < $this->max) {//连接数没达到最大，新建连接入池
76.  
    $this->count++;
77.  
    $obj = $this->createObject();
78.  
    } else {
79.  
    $obj = $this->connections->pop($timeOut);//timeout为出队的最大的等待时间
80.  
    }
81.  
    } else {
82.  
    $obj = $this->connections->pop($timeOut);
83.  
    }
84.  
    return $obj;
85.  
    }
86.  
     
87.  
    public function free($obj)
88.  
    {
89.  
    if ($obj) {
90.  
    $this->connections->push($obj);
91.  
    }
92.  
    }
93.  
     
94.  
    /**
95.  
    * 处理空闲连接
96.  
    */
97.  
    public function gcSpareObject()
98.  
    {
99.  
    //大约2分钟检测一次连接
100.  
     swoole_timer_tick(120000, function () {
101.  
     $list = [];
102.  
     /*echo "开始检测回收空闲链接" . $this->connections->length() . PHP_EOL;*/
103.  
     if ($this->connections->length() < intval($this->max * 0.5)) {
104.  
     echo "请求连接数还比较多，暂不回收空闲连接 ";
105.  
     }#1
106.  
     while (true) {
107.  
     if (!$this->connections->isEmpty()) {
108.  
     $obj = $this->connections->pop(0.001);
109.  
     $last_used_time = $obj[‘last_used_time‘];
110.  
     if ($this->count > $this->min && (time() - $last_used_time > $this->spareTime)) {//回收
111.  
     $this->count--;
112.  
     } else {
113.  
     array_push($list, $obj);
114.  
     }
115.  
     } else {
116.  
     break;
117.  
     }
118.  
     }
119.  
     foreach ($list as $item) {
120.  
     $this->connections->push($item);
121.  
     }
122.  
     unset($list);
123.  
     });
124.  
     }
125.  
     }
126.  
      

• 同步PDO客户端下实现

1.  
   <?php
2.  
   /**
3.  
   * 数据库连接池PDO方式
4.  
   * User: user
5.  
   * Date: 2018/9/8
6.  
   * Time: 11:30
7.  
   */
8.  
   require "AbstractPool.php";
9.  
    
10.  
    class MysqlPoolPdo extends AbstractPool
11.  
    {
12.  
    protected $dbConfig = array(
13.  
    ‘host‘ => ‘mysql:host=10.0.2.2:3306;dbname=test‘,
14.  
    ‘port‘ => 3306,
15.  
    ‘user‘ => ‘root‘,
16.  
    ‘password‘ => ‘root‘,
17.  
    ‘database‘ => ‘test‘,
18.  
    ‘charset‘ => ‘utf8‘,
19.  
    ‘timeout‘ => 2,
20.  
    );
21.  
    public static $instance;
22.  
     
23.  
    public static function getInstance()
24.  
    {
25.  
    if (is_null(self::$instance)) {
26.  
    self::$instance = new MysqlPoolPdo();
27.  
    }
28.  
    return self::$instance;
29.  
    }
30.  
     
31.  
    protected function createDb()
32.  
    {
33.  
    return new PDO($this->dbConfig[‘host‘], $this->dbConfig[‘user‘], $this->dbConfig[‘password‘]);
34.  
    }
35.  
    }
36.  
     
37.  
    $httpServer = new swoole_http_server(‘0.0.0.0‘, 9501);
38.  
    $httpServer->set(
39.  
    [‘worker_num‘ => 1]
40.  
    );
41.  
    $httpServer->on("WorkerStart", function () {
42.  
    MysqlPoolPdo::getInstance()->init();
43.  
    });
44.  
    $httpServer->on("request", function ($request, $response) {
45.  
    $db = null;
46.  
    $obj = MysqlPoolPdo::getInstance()->getConnection();
47.  
    if (!empty($obj)) {
48.  
    $db = $obj ? $obj[‘db‘] : null;
49.  
    }
50.  
    if ($db) {
51.  
    $db->query("select sleep(2)");
52.  
    $ret = $db->query("select * from guestbook limit 1");
53.  
    MysqlPoolPdo::getInstance()->free($obj);
54.  
    $response->end(json_encode($ret));
55.  
    }
56.  
    });
57.  
    $httpServer->start();

代码调用过程详解：
1、server启动时，调用init()方法初始化最少数量(min指定)的连接对象，放进类型为channelle的connections对象中。在init中循环调用中，依赖了createObject()返回连接对象，而createObject()
中是调用了本来实现的抽象方法，初始化返回一个PDO db连接。所以此时，连接池connections中有min个对象。

2、server监听用户请求，当接收发请求时，调用连接数的getConnection()方法从connections通道中pop()一个对象。此时如果并发了10个请求，server因为配置了1个worker,所以再pop到一个对象返回时，遇到sleep()的查询，因为用的连接对象是pdo的查询，此时的woker进程只能等待，完成后才能进入下一个请求。因此，池中的其余连接其实是多余的，同步客户端的请求速度只能和woker的数量有关。
3、查询结束后，调用free()方法把连接对象放回connections池中。

ab -c 10 -n 10运行的结果，单个worker处理，select sleep(2) 查询睡眠2s,同步客户端方式总共运行时间为20s以上，而且mysql的连接始终维持在一条。结果如下：

• 协程客户端CoroutineMySQL方式的调用

1.  
   <?php
2.  
   /**
3.  
   * 数据库连接池协程方式
4.  
   * User: user
5.  
   * Date: 2018/9/8
6.  
   * Time: 11:30
7.  
   */
8.  
   require "AbstractPool.php";
9.  
    
10.  
    class MysqlPoolCoroutine extends AbstractPool
11.  
    {
12.  
    protected $dbConfig = array(
13.  
    ‘host‘ => ‘10.0.2.2‘,
14.  
    ‘port‘ => 3306,
15.  
    ‘user‘ => ‘root‘,
16.  
    ‘password‘ => ‘root‘,
17.  
    ‘database‘ => ‘test‘,
18.  
    ‘charset‘ => ‘utf8‘,
19.  
    ‘timeout‘ => 10,
20.  
    );
21.  
    public static $instance;
22.  
     
23.  
    public static function getInstance()
24.  
    {
25.  
    if (is_null(self::$instance)) {
26.  
    self::$instance = new MysqlPoolCoroutine();
27.  
    }
28.  
    return self::$instance;
29.  
    }
30.  
     
31.  
    protected function createDb()
32.  
    {
33.  
    $db = new SwooleCoroutineMysql();
34.  
    $db->connect(
35.  
    $this->dbConfig
36.  
    );
37.  
    return $db;
38.  
    }
39.  
    }
40.  
     
41.  
    $httpServer = new swoole_http_server(‘0.0.0.0‘, 9501);
42.  
    $httpServer->set(
43.  
    [‘worker_num‘ => 1]
44.  
    );
45.  
    $httpServer->on("WorkerStart", function () {
46.  
    //MysqlPoolCoroutine::getInstance()->init()->gcSpareObject();
47.  
    MysqlPoolCoroutine::getInstance()->init();
48.  
    });
49.  
     
50.  
    $httpServer->on("request", function ($request, $response) {
51.  
    $db = null;
52.  
    $obj = MysqlPoolCoroutine::getInstance()->getConnection();
53.  
    if (!empty($obj)) {
54.  
    $db = $obj ? $obj[‘db‘] : null;
55.  
    }
56.  
    if ($db) {
57.  
    $db->query("select sleep(2)");
58.  
    $ret = $db->query("select * from guestbook limit 1");
59.  
    MysqlPoolCoroutine::getInstance()->free($obj);
60.  
    $response->end(json_encode($ret));
61.  
    }
62.  
    });
63.  
    $httpServer->start();

代码调用过程详解
1、同样的，协程客户端方式下的调用，也是实现了之前封装好的连接池类AbstractPool.php。只是createDb()的抽象方法用了swoole内置的协程客户端去实现。
2、server启动后，初始化都和同步一样。不一样的在获取连接对象的时候，此时如果并发了10个请求，同样是配置了1个worker进程在处理，但是在第一请求到达，pop出池中的一个连接对象，执行到query()方法，遇上sleep阻塞时，此时，woker进程不是在等待select的完成，而是切换到另外的协程去处理下一个请求。完成后同样释放对象到池中。当中有重点解释的代码段中getConnection()中。

1.  
   public function getConnection($timeOut = 3)
2.  
   {
3.  
   $obj = null;
4.  
   if ($this->connections->isEmpty()) {
5.  
   if ($this->count < $this->max) {//连接数没达到最大，新建连接入池
6.  
   $this->count++;
7.  
   $obj = $this->createObject();#1
8.  
   } else {
9.  
   $obj = $this->connections->pop($timeOut);#2
10.  
    }
11.  
    } else {
12.  
    $obj = $this->connections->pop($timeOut);#3
13.  
    }
14.  
    return $obj;
15.  
    }

当调用到getConnection()时，如果此时由于大量并发请求过多，连接池connections为空，而没达到最大连接max数量时时，代码运行到#1处，调用了createObject(),新建连接返回；但如果连接池connections为空，而到达了最大连接数max时，代码运行到了#2处，也就是$this->connections->pop($timeOut)，此时会阻塞$timeOut的时间，如果期间有链接释放了，会成功获取到，然后协程返回。超时没获取到，则返回false。

3、最后说一下协程Mysql客户端一项重要配置，那就是代码里$dbConfig中timeout值的配置。这个配置是意思是最长的查询等待时间。可以看一个例子说明下：

1.  
   go(function () {
2.  
   $start = microtime(true);
3.  
   $db = new SwooleCoroutineMySQL();
4.  
   $db->connect([
5.  
   ‘host‘ => ‘10.0.2.2‘,
6.  
   ‘port‘ => 3306,
7.  
   ‘user‘ => ‘root‘,
8.  
   ‘password‘ => ‘root‘,
9.  
   ‘database‘ => ‘test‘,
10.  
    ‘timeout‘ => 4#1
11.  
    ]);
12.  
    $db->query("select sleep(5)");
13.  
    echo "我是第一个sleep五秒之后 ";
14.  
    $ret = $db->query("select user from guestbook limit 1");#2
15.  
    var_dump($ret);
16.  
    $use = microtime(true) - $start;
17.  
    echo "协程mysql输出用时：" . $use . PHP_EOL;
18.  
    });

#1处代码，如果timeout配了4s查询超时，而第一条查询select sleep(5)阻塞后，协程切换到下一条sql的执行，其实$db并不能执行成功，因为用一个连接，同一个协程中，其实执行是同步的，所以此时第二条查询在等待4s超时后，没获取到db的连接执行，就会执行失败。而如果第一条查询执行的时间少于这个timeout,那么会执行查询成功。猜猜上面执行用时多少？结果如下：

如果把timeout换成6s呢，结果如下：

所以要注意的是，协程的客户端内执行其实是同步的，不要理解为异步，它只是遇到IO阻塞时能让出执行权，切换到其他协程而已，不能和异步混淆。

ab -c 10 -n 10运行的结果，单个worker处理，select sleep(2) 查询睡眠2s,协程客户端方式总共运行时间为2s多。结果如下：

数据库此时的连接数为10条（show full PROCESSLIST）：

再尝试 ab -c 200 -n 1000 http://127.0.0.1:9501/，200多个并发的处理，时间是20多秒，mysql连接数达到指定的最大值100个。结果如下：

 

四、后言

现在连接池基本实现了高并发时的连接分配和控制，但是还有一些细节要处理，例如：

• 并发时，建立了max个池对象，不能一直在池中维护这么多，要在请求空闲时，把连接池的数量维持在一个空闲值内。这里是简单做了gcSpareObject()的方法实现空闲处理。直接在初始化woker的时候调用：MysqlPoolCoroutine::getInstance()->init()->gcSpareObject();就会定时检测回收。问题是如何判断程序比较空闲，值得再去优化。
• 定时检测连接时候是活的，剔除死链
• 假如程序忘记调用free()释放对象到池，是否有更好方法避免这种情况？

对于以上，希望各大神看到后，能提供不错的意见！

转载于:https://my.oschina.net/u/2394701/blog/2046414