1. <?php
    2. /*
    3.  * 接收客户端的sql,执行sql,结果返回
    4.  * */
    5. class DBServer{
    7.     protected  $server;
    8.     protected  $min; //最小连接数
    9.     protected  $max; //最大连接数
    10.     protected  $idle_pool;//空闲连接池
    11.     protected  $worker_pool;//工作连接池
    12.     protected  $wait_queue;//工作连接池
    13.     protected  $db_config; //配置信息
    14.     public  function  __construct($config)
    15.     {
    16.         //全局对象
    17.         $this->server=new \swoole\server('0.0.0.0',9800);
    18.         $this->server->set($config['tcp']);
    19.         //注册事件
    20.         $this->server->on('WorkerStart',[$this,'onWorkerStart']);
    21.         $this->server->on('receive',[$this,'onReceive']);
    22.         //$this->server->on('close',[$this,'onClose']);
    23.         $this->max=$config['server']['max'];
    24.         $this->min=$config['server']['min'];
    25.         $this->idle_pool=new SplQueue();
    26.        // $this->worker_pool=new SplQueue();
    27.         //$this->wait_queue=new SplQueue();
    28.         $this->db_config=$config;
    29.         //$this->idle_pool->push('1');
    30.         //$this->idle_pool->push('2');
    31.         //echo $this->idle_pool->shift();
    32.         $this->server->start(); //全局生命周期
    33.     }
    35.     public  function  onWorkerStart($server,$worker_id){
    37.         //生成数据库连接
    38.         for ($i=0;$i<$this->min;$i++){
    39.             $pdo=$this->db_connect($this->db_config);
    40.             $this->push_queue('idle_pool',$pdo);
    41.         }
    43.        /*swoole_timer_tick(1000,function (){
    44.                  //var_dump($this->wait_queue);
    45.         });*/
    49.     }
    51.     public  function  onReceive($server,$fd,$reactor_id,$data){
    52.         $len=unpack('N',$data)[1];
    53.         $body=substr($data,-$len);//去除二进制数据之后,不要包头的数据
    55.         //判断有没有空闲连接,等待sql执行
    56.         /*if(count( $this->idle_pool)===0){
    57.             $queueName='wait_queue';
    58.             $this->push_queue($queueName,$body,$fd);
    59.         }*/
    61.         $res=$this->query($fd,$body); //执行一条sql
    62.         var_dump(count($res));
    63.     }
    65.     protected  function query($fd,$sql){
    66.         $idle_pool=$this->idle_pool->shift();
    67.         $db=$idle_pool['db'];
    68.         //执行sql的时候,两次循环,如果执行成功,就跳出循环失败
    70.         for ($i=0;$i<2;$i++){
    71.             try{
    72.                 $res=$db->query($sql);
    73.                 break;
    74.             }catch (PDOException $e){
    75.                 $err_msg=$e->getMessage();
    76.                 if(strpos($err_msg,'MySQL server has gone away')!==false){
    77.                        $db=$this->db_connect($this->db_config);
    78.                 }
    79.                 $res=$db->query($sql);
    80.                 break;
    81.             }
    83.         }
    85.         //如果当前是select
    86.         if(preg_match('/^select/i',$sql)){
    87.             $data=$res->fetchAll(PDO::FETCH_ASSOC); //执行查询
    88.         }else{
    89.             $data=$res;
    90.         }
    91.         //sleep(2); //同步执行,如果有sql没有执行完
    92.         //执行完毕,重新放回到空闲连接池
    93.         $this->push_queue('idle_pool',$db,$fd);
    95.         return $data;
    96.     }
    98.     //向队列当中添加连接
    99.     public  function  push_queue($queueName,$pdo,$fd=0){
    100.         $arr=[
    101.             'fd'=>$fd,
    102.             'db'=>$pdo
    103.         ];
    104.         $this->$queueName->push($arr);
    105.     }
    108.     /**
    109.      * 数据库长连接
    110.      * @param $db_config
    111.      * @return array|PDO
    112.      */
    113.     public  function  db_connect($db_config){
    114.          try{
    115.              $pdo=new PDO($db_config['db']['dsn'],$db_config['db']['user'],$db_config['db']['password'],[PDO::ATTR_PERSISTENT=>true,PDO::ATTR_SERVER_INFO=>true]);
    116.              return $pdo;
    117.          }catch (Exception $e){
    118.             echo $e->getMessage();
    119.             return [];
    120.          }
    121.     }
    123. }
    125. $config['tcp']=[
    126.     'worker_num'=>1,
    127.     'open_length_check' => true,
    128.     'package_length_type'=>'N',
    129.     'package_length_offset'=>0, //计算总长度
    130.     'package_body_offset'=>4,//包体位置
    131.     ];
    133. $config['server']=[
    134.     'max'=>20,
    135.     'min'=>2
    136. ];
    138. $config['db']=[
    139.     'dsn'=>'mysql:host=localhost;dbname=test',
    140.     'user'=>'root',
    141.     'password'=>'Qq!990979940'
    142. ];
    144. new DBServer($config);