目录

这是关于 Swoole 入门学习的第八篇文章：Swoole MySQL 连接池的实现。

收到读者的咨询，这情况大家可能也会有，所以就在这说说：

“亮哥，我今年30岁了，有点中年危机，最近有点焦虑，发现工作虽然很忙，但是没感觉能力有所提升，整天都有写不完的业务代码，时间紧有时代码质量也不怎么样，知道还有很多改进空间，但一直没时间改，主要是后面项目压着，马上又要进入开发了，这种情况怎么办？”

首先，我是菜鸡，观点不喜勿喷，那我就说下自己的看法：

上面的描述比较主观，人呀有时候发现不了自己的能力很正常，有时候有能力了并不是马上就能显现的，而是到了某个阶段后突然发现，哇塞，原来自己这么厉害。

当然能力也分很多种，比如专业能力，快速学习能力，进度把控能力，还有自信也是一种能力，不要脸是一种能力，坚持不要脸更是一种能力。

其实能力提升最快的还是靠工作实践，悄悄问问自己加入了很多大牛的微信群，能力提升了吗？看书自学不实践是不是吸收的也不多。

如果非要给一个具体的方案，那就是在团队内多分享吧，因为在分享前你会做充分的准备来避免分享时出丑，即使有时候自己知道，当讲出来的时候就不是那么回事了。

前期分享可以是看稿，后期练习无稿分享。

然后，再多说一点，30了给自己一个目标，不要盲目每天就是学学学，比如目标是技术专家，目标是业务专家，都很好呀，当然目标与自己性格有关也不是一成不变的。

围绕着目标设置一些计划，不要以为每天的学学学，就觉得其他的一切就自然而来，其中还有很多机遇和人脉的因素。

最后，如果实在感觉压得喘不过气，就换个环境吧，别和自己过不去。

开始今天的文章，这篇文章实现了 Swoole MySQL 连接池，代码是在《Swoole RPC 的实现》文章的基础上进行开发的。

先回顾上篇文章的内容：

  • 实现了 HTTP / TCP 请求
  • 实现了 同步 / 异步 请求
  • 分享了 OnRequest.php、OnReceive.php 源码
  • 业务逻辑 Order.php 中返回的是假数据

本篇文章主要的功能点：

  • 业务逻辑 Order.php 中返回 MySQL 数据库中的数据。
  • Task 启用了协程
  • 支持 主/从 数据库配置
  • 实现数据库连接池
  • 实现数据库 CURD
  1. <?php
  3. if (!defined(\'SERVER_PATH\')) exit("No Access");
  5. class Order
  6. {
  7.     public function get_list($uid = 0, $type = 0)
  8.     {
  9.         //TODO 业务代码
  11.         $rs[0][\'order_code\'] = \'1\';
  12.         $rs[0][\'order_name\'] = \'订单1\';
  14.         $rs[1][\'order_code\'] = \'2\';
  15.         $rs[1][\'order_name\'] = \'订单2\';
  17.         $rs[2][\'order_code\'] = \'3\';
  18.         $rs[2][\'order_name\'] = \'订单3\';
  20.         return $rs;
  21.     }
  22. }
  24. 修改成：
  26. class Order
  27. {
  28.     private $mysql;
  29.     private $table;
  31.     public function __construct()
  32.     {
  33.         $pool = MysqlPool::getInstance();
  34.         $this->mysql = $pool->get();
  35.         $this->table = \'order\';
  36.     }
  38.     public function add($code = \'\', $name = \'\')
  39.     {
  40.         //TODO 验证
  41.         return $this->mysql->insert($this->table, [\'code\' => $code, \'name\' => $name]);
  42.     }
  44.     public function edit($id = 0,  $name=\'\')
  45.     {
  46.         //TODO 验证
  47.         return $this->mysql->update($this->table, [\'name\' => $name], [\'id\' => $id]);
  48.     }
  50.     public function del($id = 0)
  51.     {
  52.         //TODO 验证
  53.         return $this->mysql->delete($this->table, [\'id\' => $id]);
  54.     }
  56.     public function info($code = \'\')
  57.     {
  58.         //TODO 验证
  59.         return $this->mysql->select($this->table, [\'code\' => $code]);
  60.     }
  61. }

一、需要新增两项配置：

  1. enable_coroutine      = true
  2. task_enable_coroutine = true

二、回调参数发生改变

  1. $serv->on(\'Task\', function ($serv, $task_id, $src_worker_id, $data) {
  2.    ...
  3. });
  5. 修改成：
  7. $serv->on(\'Task\', function ($serv, $task) {
  8.     $task->worker_id; //来自哪个`Worker`进程
  9.     $task->id; //任务的编号
  10.     $task->data; //任务的数据
  11. });

Mysql.php

  1. <?php
  3. if (!defined(\'SERVER_PATH\')) exit("No Access");
  5. $db[\'default\'][\'pool_size\']        = 3; //连接池个数
  6. $db[\'default\'][\'pool_get_timeout\'] = 0.5; //获取连接池超时时间
  7. $db[\'default\'][\'timeout\']          = 0.5; //数据库建立连接超时时间
  8. $db[\'default\'][\'charset\']          = \'utf8\'; //字符集
  9. $db[\'default\'][\'strict_type\']      = false; //开启严格模式
  10. $db[\'default\'][\'fetch_mode\']       = true; //开启fetch模式
  12. $config[\'master\']             = $db[\'default\'];
  13. $config[\'master\'][\'host\']     = \'127.0.0.1\';
  14. $config[\'master\'][\'port\']     = 3306;
  15. $config[\'master\'][\'user\']     = \'root\';
  16. $config[\'master\'][\'password\'] = \'123456\';
  17. $config[\'master\'][\'database\'] = \'demo\';
  19. $config[\'slave\']             = $db[\'default\'];
  20. $config[\'slave\'][\'host\']     = \'127.0.0.1\';
  21. $config[\'slave\'][\'port\']     = 3306;
  22. $config[\'slave\'][\'user\']     = \'root\';
  23. $config[\'slave\'][\'password\'] = \'123456\';
  24. $config[\'slave\'][\'database\'] = \'demo\';

MysqlPool.php

  1. <?php
  3. if (!defined(\'SERVER_PATH\')) exit("No Access");
  5. class MysqlPool
  6. {
  7.     private static $instance;
  8.     private $pool;
  9.     private $config;
  11.     public static function getInstance($config = null)
  12.     {
  13.         if (empty(self::$instance)) {
  14.             if (empty($config)) {
  15.                 throw new RuntimeException("MySQL config empty");
  16.             }
  17.             self::$instance = new static($config);
  18.         }
  19.         return self::$instance;
  20.     }
  22.     public function __construct($config)
  23.     {
  24.         if (empty($this->pool)) {
  25.             $this->config = $config;
  26.             $this->pool = new chan($config[\'master\'][\'pool_size\']);
  27.             for ($i = 0; $i < $config[\'master\'][\'pool_size\']; $i++) {
  28.                 go(function() use ($config) {
  29.                     $mysql = new MysqlDB();
  30.                     $res = $mysql->connect($config);
  31.                     if ($res === false) {
  32.                         throw new RuntimeException("Failed to connect mysql server");
  33.                     } else {
  34.                         $this->pool->push($mysql);
  35.                     }
  36.                 });
  37.             }
  38.         }
  39.     }
  41.     public function get()
  42.     {
  43.         if ($this->pool->length() > 0) {
  44.             $mysql = $this->pool->pop($this->config[\'master\'][\'pool_get_timeout\']);
  45.             if (false === $mysql) {
  46.                 throw new RuntimeException("Pop mysql timeout");
  47.             }
  48.             defer(function () use ($mysql) { //释放
  49.                 $this->pool->push($mysql);
  50.             });
  51.             return $mysql;
  52.         } else {
  53.             throw new RuntimeException("Pool length <= 0");
  54.         }
  55.     }
  56. }

MysqlDB.php

  1. <?php
  3. if (!defined(\'SERVER_PATH\')) exit("No Access");
  5. class MysqlDB
  6. {
  7.     private $master;
  8.     private $slave;
  9.     private $config;
  11.     public function __call($name, $arguments)
  12.     {
  13.         if ($name != \'query\') {
  14.             throw new RuntimeException($name."：This command is not supported");
  15.         } else {
  16.             return $this->_execute($arguments[0]);
  17.         }
  18.     }
  20.     public function connect($config)
  21.     {
  22.         //主库
  23.         $master = new Swoole\Coroutine\MySQL();
  24.         $res = $master->connect($config[\'master\']);
  25.         if ($res === false) {
  26.             throw new RuntimeException($master->connect_error, $master->errno);
  27.         } else {
  28.             $this->master = $master;
  29.         }
  31.         //从库
  32.         $slave = new Swoole\Coroutine\MySQL();
  33.         $res = $slave->connect($config[\'slave\']);
  34.         if ($res === false) {
  35.             throw new RuntimeException($slave->connect_error, $slave->errno);
  36.         } else {
  37.             $this->slave = $slave;
  38.         }
  40.         $this->config = $config;
  41.         return $res;
  42.     }
  44.     public function insert($table = \'\', $data = [])
  45.     {
  46.         $fields = \'\';
  47.         $values = \'\';
  48.         $keys = array_keys($data);
  49.         foreach ($keys as $k) {
  50.             $fields .= "`".addslashes($k)."`, ";
  51.             $values .= "\'".addslashes($data[$k])."\', ";
  52.         }
  53.         $fields = substr($fields, 0, -2);
  54.         $values = substr($values, 0, -2);
  55.         $sql = "INSERT INTO `{$table}` ({$fields}) VALUES ({$values})";
  56.         return $this->_execute($sql);
  57.     }
  59.     public function update($table = \'\', $set = [], $where = [])
  60.     {
  61.         $arr_set = [];
  62.         foreach ($set as $k => $v) {
  63.             $arr_set[] = \'`\'.$k . \'` = \' . $this->_escape($v);
  64.         }
  65.         $set = implode(\', \', $arr_set);
  66.         $where = $this->_where($where);
  67.         $sql = "UPDATE `{$table}` SET {$set} {$where}";
  68.         return $this->_execute($sql);
  69.     }
  71.     public function delete($table = \'\', $where = [])
  72.     {
  73.         $where = $this->_where($where);
  74.         $sql = "DELETE FROM `{$table}` {$where}";
  75.         return $this->_execute($sql);
  76.     }
  78.     public function select($table = \'\',$where = [])
  79.     {
  80.         $where = $this->_where($where);
  81.         $sql = "SELECT * FROM `{$table}` {$where}";
  82.         return $this->_execute($sql);
  83.     }
  85.     private function _where($where = [])
  86.     {
  87.         $str_where = \'\';
  88.         foreach ($where as $k => $v) {
  89.             $str_where .= " AND `{$k}` = ".$this->_escape($v);
  90.         }
  91.         return "WHERE 1 ".$str_where;
  92.     }
  94.     private function _escape($str)
  95.     {
  96.         if (is_string($str)) {
  97.             $str = "\'".$str."\'";
  98.         } elseif (is_bool($str)) {
  99.             $str = ($str === FALSE) ? 0 : 1;
  100.         } elseif (is_null($str)) {
  101.             $str = \'NULL\';
  102.         }
  103.         return $str;
  104.     }
  106.     private function _execute($sql)
  107.     {
  108.         if (strtolower(substr($sql, 0, 6)) == \'select\') {
  109.             $db = $this->_get_usable_db(\'slave\');
  110.         } else {
  111.             $db = $this->_get_usable_db(\'master\');
  112.         }
  113.         $result = $db->query($sql);
  114.         if ($result === true) {
  115.             return [
  116.                 \'affected_rows\' => $db->affected_rows,
  117.                 \'insert_id\'     => $db->insert_id,
  118.             ];
  119.         }
  120.         return $result;
  121.     }
  123.     private function _get_usable_db($type)
  124.     {
  125.         if ($type == \'master\') {
  126.             if (!$this->master->connected) {
  127.                 $master = new Swoole\Coroutine\MySQL();
  128.                 $res = $master->connect($this->config[\'master\']);
  129.                 if ($res === false) {
  130.                     throw new RuntimeException($master->connect_error, $master->errno);
  131.                 } else {
  132.                     $this->master = $master;
  133.                 }
  134.             }
  135.             return $this->master;
  136.         } elseif ($type == \'slave\') {
  137.             if (!$this->slave->connected) {
  138.                 $slave = new Swoole\Coroutine\MySQL();
  139.                 $res = $slave->connect($this->config[\'slave\']);
  140.                 if ($res === false) {
  141.                     throw new RuntimeException($slave->connect_error, $slave->errno);
  142.                 } else {
  143.                     $this->slave = $slave;
  144.                 }
  145.             }
  146.             return $this->slave;
  147.         }
  148.     }
  149. }
  1. try {
  2.     MysqlPool::getInstance(get_config(\'mysql\'));
  3. } catch (\Exception $e) {
  4.     $serv->shutdown();
  5. } catch (\Throwable $throwable) {
  6.     $serv->shutdown();
  7. }
  1. <?php
  3. //新增
  4. $demo = [
  5.     \'type\'  => \'SW\',
  6.     \'token\' => \'Bb1R3YLipbkTp5p0\',
  7.     \'param\' => [
  8.         \'class\'  => \'Order\',
  9.         \'method\' => \'add\',
  10.         \'param\' => [
  11.             \'code\' => \'C\'.mt_rand(1000,9999),
  12.             \'name\' => \'订单-\'.mt_rand(1000,9999),
  13.         ],
  14.     ],
  15. ];
  17. //编辑
  18. $demo = [
  19.     \'type\'  => \'SW\',
  20.     \'token\' => \'Bb1R3YLipbkTp5p0\',
  21.     \'param\' => [
  22.         \'class\'  => \'Order\',
  23.         \'method\' => \'edit\',
  24.         \'param\' => [
  25.             \'id\' => \'4\',
  26.             \'name\' => \'订单-\'.mt_rand(1000,9999),
  27.         ],
  28.     ],
  29. ];
  31. //删除
  32. $demo = [
  33.     \'type\'  => \'SW\',
  34.     \'token\' => \'Bb1R3YLipbkTp5p0\',
  35.     \'param\' => [
  36.         \'class\'  => \'Order\',
  37.         \'method\' => \'del\',
  38.         \'param\' => [
  39.             \'id\' => \'1\',
  40.         ],
  41.     ],
  42. ];
  45. //查询
  46. $demo = [
  47.     \'type\'  => \'SW\',
  48.     \'token\' => \'Bb1R3YLipbkTp5p0\',
  49.     \'param\' => [
  50.         \'class\'  => \'Order\',
  51.         \'method\' => \'info\',
  52.         \'param\' => [
  53.             \'code\' => \'C4649\'
  54.         ],
  55.     ],
  56. ];
  58. $ch = curl_init();
  59. $options = [
  60.     CURLOPT_URL  => \'http://10.211.55.4:9509/\',
  61.     CURLOPT_POST => 1,
  62.     CURLOPT_POSTFIELDS => json_encode($demo),
  63. ];
  64. curl_setopt_array($ch, $options);
  65. curl_exec($ch);
  66. curl_close($ch);

官方协程 MySQL 客户端手册：

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

大家可以尝试使用官方提供的其他方法。

Demo 代码仅供参考，里面有很多不严谨的地方。

根据自己的需要进行修改，比如业务代码相关验证，CURD 方法封装 …

推荐一个完善的产品，Swoole 开发的 MySQL 数据库连接池（SMProxy）：

https://github.com/louislivi/smproxy

上面的 Demo 需要源码的，加我微信。（菜单-> 加我微信-> 扫我）

本文欢迎转发，转发请注明作者和出处，谢谢！

版权声明：本文为xinliangcoder原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://www.cnblogs.com/xinliangcoder/p/10930035.html