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

基于Swoole扩展开发异步高性能的MySQL代理服务器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中取出等待的请求进行处理。如此循环使用。由于整个服务器是异步的单进程单线程所以完全不需要锁。而且是完全异步的，效率非常高。当然本文的代码，如果要用于生产环境，还需做更多的保护机制和压力测试。在此仅抛砖引玉，提供一个解决问题的思路。classDBServer{protected$pool_size=20;protected$idle_pool=array();//空闲连接protected$busy_pool=array();//工作连接protected$wait_queue=array();//等待的请求protected$wait_queue_max=100;//等待队列的最大长度，超过后将拒绝新的请求/***@varswoole_server*/protected$serv;functionrun(){$serv=newswoole_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();}functiononStart($serv){$this-serv=$serv;for($i=0;$i$this-pool_size;$i++){$db=newmysqli;$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,);}echoServer:start.Swooleversionis[.SWOOLE_VERSION.]\n;}functiononSQLReady($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(MySQLiError:%s\n,mysqli_error($mysqli)));}//releasemysqliobject$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']);}}}functiononReceive($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,requesttoomany,Pleasetryagainlater.);}}else{$this-doQuery($fd,$data);}}functiondoQuery($fd,$sql){//从空闲池中移除$db=array_pop($this-idle_pool);/***@varmysqli*/$mysqli=$db['mysqli'];for($i=0;$i2;$i++){$result=$mysqli-query($sql,MYSQLI_ASYNC);if($result===false){if($mysqli-errno==2013or$mysqli-errno==2006){$mysqli-close();$r=$mysqli-connect();if($r===true)continue;}}break;}$db['fd']=$fd;//加入工作池中$this-busy_pool[$db['db_sock']]=$db;}}$server=newDBServer();$server-run();