基于Swoole扩展开发异步高性能的MySQL代理服务器

MySQL数据库对每个客户端连接都会分配一个线程&＃xff0c;所以连接非常宝贵。开发一个异步的MySQL代理服务器&＃xff0c;PHP应用服务器可以长连接到这台Server&＃xff0c;既减轻MYSQL的连接压力&＃xff0c;又使PHP保持长连接减少connect/close的网络开销。

此Server考虑到了设置了数据库连接池尺寸&＃xff0c;区分忙闲&＃xff0c;mysqli断线重连&＃xff0c;并设置了负载保护。基于swoole扩展开发&＃xff0c;io循环使用epoll&＃xff0c;是全异步非阻塞的&＃xff0c;可以应对大量TCP连接。

程序的逻辑是&＃xff1a;启动时创建N个MySQL连接&＃xff0c;收到客户端发来的SQL后&＃xff0c;分配1个MySQL连接&＃xff0c;将SQL发往数据库服务器。然后等待数据库返回查询结果。当数据库返回结果后&＃xff0c;再发给对应的客户端连接。

核心的数据结构是3个PHP数组。idle_pool是空闲的数据库连接&＃xff0c;当有SQL请求时从idle_pool中移到busy_pool中。当数据库返回结果后从busy_pool中再移到idle_pool中&＃xff0c;以供新的请求使用。当SQL请求到达时如果没有空闲的数据库连接&＃xff0c;那会自动加入到wait_queue中。一旦有SQL完成操作&＃xff0c;将自动从wait_queue中取出等待的请求进行处理。

如此循环使用。由于整个服务器是异步的单进程单线程所以完全不需要锁。而且是完全异步的&＃xff0c;效率非常高。

当然本文的代码&＃xff0c;如果要用于生产环境&＃xff0c;还需做更多的保护机制和压力测试。在此仅抛砖引玉&＃xff0c;提供一个解决问题的思路。

class DBServer
{protected $pool_size &＃61; 20;protected $idle_pool &＃61; array(); //空闲连接protected $busy_pool &＃61; array(); //工作连接protected $wait_queue &＃61; array(); //等待的请求protected $wait_queue_max &＃61; 100; //等待队列的最大长度&＃xff0c;超过后将拒绝新的请求/*** &＃64;var swoole_server*/protected $serv;function run(){$serv &＃61; new swoole_server("127.0.0.1", 9509);$serv->set(array(&＃39;worker_num&＃39; &＃61;> 1,));$serv->on(&＃39;WorkerStart&＃39;, array($this, &＃39;onStart&＃39;));//$serv->on(&＃39;Connect&＃39;, array($this, &＃39;onConnect&＃39;));$serv->on(&＃39;Receive&＃39;, array($this, &＃39;onReceive&＃39;));//$serv->on(&＃39;Close&＃39;, array($this, &＃39;onClose&＃39;));$serv->start();}function onStart($serv){$this->serv &＃61; $serv;for ($i &＃61; 0; $i <$this->pool_size; $i&＃43;&＃43;) {$db &＃61; new mysqli;$db->connect(&＃39;127.0.0.1&＃39;, &＃39;root&＃39;, &＃39;root&＃39;, &＃39;test&＃39;);$db_sock &＃61; swoole_get_mysqli_sock($db);swoole_event_add($db_sock, array($this, &＃39;onSQLReady&＃39;));$this->idle_pool[] &＃61; array(&＃39;mysqli&＃39; &＃61;> $db,&＃39;db_sock&＃39; &＃61;> $db_sock,&＃39;fd&＃39; &＃61;> 0,);}echo "Server: start.Swoole version is [" . SWOOLE_VERSION . "]\n";}function onSQLReady($db_sock){$db_res &＃61; $this->busy_pool[$db_sock];$mysqli &＃61; $db_res[&＃39;mysqli&＃39;];$fd &＃61; $db_res[&＃39;fd&＃39;];echo __METHOD__ . ": client_sock&＃61;$fd|db_sock&＃61;$db_sock\n";if ($result &＃61; $mysqli->reap_async_query()) {$ret &＃61; var_export($result->fetch_all(MYSQLI_ASSOC), true) . "\n";$this->serv->send($fd, $ret);if (is_object($result)) {mysqli_free_result($result);}} else {$this->serv->send($fd, sprintf("MySQLi Error: %s\n", mysqli_error($mysqli)));}//release mysqli object$this->idle_pool[] &＃61; $db_res;unset($this->busy_pool[$db_sock]);//这里可以取出一个等待请求if (count($this->wait_queue) > 0) {$idle_n &＃61; count($this->idle_pool);for ($i &＃61; 0; $i <$idle_n; $i&＃43;&＃43;) {$req &＃61; array_shift($this->wait_queue);$this->doQuery($req[&＃39;fd&＃39;], $req[&＃39;sql&＃39;]);}}}function onReceive($serv, $fd, $from_id, $data){//没有空闲的数据库连接if (count($this->idle_pool) &＃61;&＃61; 0) {//等待队列未满if (count($this->wait_queue) <$this->wait_queue_max) {$this->wait_queue[] &＃61; array(&＃39;fd&＃39; &＃61;> $fd,&＃39;sql&＃39; &＃61;> $data,);} else {$this->serv->send($fd, "request too many, Please try again later.");}} else {$this->doQuery($fd, $data);}}function doQuery($fd, $sql){//从空闲池中移除$db &＃61; array_pop($this->idle_pool);/*** &＃64;var mysqli*/$mysqli &＃61; $db[&＃39;mysqli&＃39;];for ($i &＃61; 0; $i <2; $i&＃43;&＃43;) {$result &＃61; $mysqli->query($sql, MYSQLI_ASYNC);if ($result &＃61;&＃61;&＃61; false) {if ($mysqli->errno &＃61;&＃61; 2013 or $mysqli->errno &＃61;&＃61; 2006) {$mysqli->close();$r &＃61; $mysqli->connect();if ($r &＃61;&＃61;&＃61; true) continue;}}break;}$db[&＃39;fd&＃39;] &＃61; $fd;//加入工作池中$this->busy_pool[$db[&＃39;db_sock&＃39;]] &＃61; $db;}
}$server &＃61; new DBServer();
$server->run();