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

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

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

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

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

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

class DBServer

{

protected $pool_size = 20;

protected $idle_pool = array(); //空闲连接

protected $busy_pool = array(); //工作连接

protected $wait_queue = array(); //等待的请求

protected $wait_queue_max = 100; //等待队列的最大长度，超过后将拒绝新的请求

/**

* @var swoole_server

*/

protected $serv;

function run()

{

$serv = new swoole_server("127.0.0.1", 9509);

$serv->set(array(

'worker_num' => 1,

));

$serv->on('WorkerStart', array($this, 'onStart'));

//$serv->on('Connect', array($this, 'onConnect'));

$serv->on('Receive', array($this, 'onReceive'));

//$serv->on('Close', array($this, 'onClose'));

$serv->start();

}

function onStart($serv)

{

$this->serv = $serv;

for ($i = 0; $i < $this->pool_size; $i++) {

$db = new mysqli;

$db->connect('127.0.0.1', 'root', 'root', 'test');

$db_sock = swoole_get_mysqli_sock($db);

swoole_event_add($db_sock, array($this, 'onSQLReady'));

$this->idle_pool[] = array(

'mysqli' => $db,

'db_sock' => $db_sock,

'fd' => 0,

);

}

echo "Server: start.Swoole version is [" . SWOOLE_VERSION . "]\n";

}

function onSQLReady($db_sock)

{

$db_res = $this->busy_pool[$db_sock];

$mysqli = $db_res['mysqli'];

$fd = $db_res['fd'];

echo __METHOD__ . ": client_sock=$fd|db_sock=$db_sock\n";

if ($result = $mysqli->reap_async_query()) {

$ret = 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[] = $db_res;

unset($this->busy_pool[$db_sock]);

//这里可以取出一个等待请求

if (count($this->wait_queue) > 0) {

$idle_n = count($this->idle_pool);

for ($i = 0; $i < $idle_n; $i++) {

$req = array_shift($this->wait_queue);

$this->doQuery($req['fd'], $req['sql']);

}

}

}

function onReceive($serv, $fd, $from_id, $data)

{

//没有空闲的数据库连接

if (count($this->idle_pool) == 0) {

//等待队列未满

if (count($this->wait_queue) < $this->wait_queue_max) {

$this->wait_queue[] = array(

'fd' => $fd,

'sql' => $data,

);

} else {

$this->serv->send($fd, "request too many, Please try again later.");

}

} else {

$this->doQuery($fd, $data);

}

}

function doQuery($fd, $sql)

{

//从空闲池中移除

$db = array_pop($this->idle_pool);

/**

* @var mysqli

*/

$mysqli = $db['mysqli'];

for ($i = 0; $i < 2; $i++) {

$result = $mysqli->query($sql, MYSQLI_ASYNC);

if ($result === false) {

if ($mysqli->errno == 2013 or $mysqli->errno == 2006) {

$mysqli->close();

$r = $mysqli->connect();

if ($r === true) continue;

}

}

break;

}

$db['fd'] = $fd;

//加入工作池中

$this->busy_pool[$db['db_sock']] = $db;

}

}

$server = new DBServer();

$server->run();

读者们如果觉得还可以就帮忙点个赞 蟹蟹

另外读者福利，可直接点击链接领取相关学习福利包：PHP高薪就业(视频、学习路线、免费获取)​www.jianshu.com

是安全网站放心，继续访问就可以领取了哦

转发开源