1. <?php
    3. /*
    4.  * 接收客户端的sql,执行sql,结果返回
    5.  * */
    6. class DBServer{
    8.     protected  $server;
    9.     protected  $min; //最小连接数
    10.     protected  $max; //最大连接数
    11.     protected  $idle_pool;//空闲连接池
    12.     protected  $worker_pool;//工作连接池
    13.     protected  $wait_queue;//等待队列
    14.     protected  $db_config; //配置信息
    15.     protected  $wait_queue_max;//最多的排队数
    16.     protected  $count; //记录当前数据库的连接数量
    17.     protected  $table;
    18.     public  function  __construct($config)
    19.     {
    20.         //全局对象
    21.         $this->server=new \swoole\server('0.0.0.0',9800);
    22.         $this->server->set($config['tcp']);
    23.         //注册事件
    24.         $this->server->on('WorkerStart',[$this,'onWorkerStart']);
    25.         $this->server->on('receive',[$this,'onReceive']);
    26.         //$this->server->on('close',[$this,'onClose']);
    27.         $this->max=$config['server']['max'];
    28.         $this->min=$config['server']['min'];
    29.         $this->count=0;
    30.         $this->idle_pool=new SplQueue();
    31.         $this->wait_queue=new SplQueue();
    32.         $this->wait_queue_max=35;
    34.         //$this->worker_pool=new SplQueue();
    35.         $this->db_config=$config;
    36.         //$this->idle_pool->push('1');
    37.         //$this->idle_pool->push('2');
    38.         //echo $this->idle_pool->shift();
    41.         $this->table=new swoole\table(1024);
    42.         //一个字段是保存的是task_worker的id号,保存的是状态
    43.         $this->table->column('count',swoole\table::TYPE_INT,5); //连接数
    44.         $this->table->create(); //必须在创建子进程之前.
    47.         $this->server->start(); //全局生命周期
    48.     }
    50.     public  function  onWorkerStart($server,$worker_id){
    52.         //生成数据库连接
    53.         for ($i=0;$i<$this->min;$i++){
    54.              $this->db_connect($this->db_config);
    55.         }
    57.         swoole_timer_tick(1000,function (){
    58.                  var_dump('创建的连接数'.$this->count.'-----'.'等待队列数'.count($this->wait_queue).'------连接池个数'.count($this->idle_pool));
    59.          });
    61.         //没有限制queue, shift为空错误,如何解决
    62.         /*
    63.         swoole_timer_tick(100,function (){
    64.              if(count($this->wait_queue)>0){
    65.                   //空闲连接池当中去取出连接去执行sql
    66.                   $idle_num=count($this->idle_pool);
    67.                   for($i=0;$i<$idle_num;$i++){
    68.                         $req=$this->wait_queue->shift();
    69.                         $this->query($req['db'],function ($res){
    70.                             //var_dump($res);
    71.                             var_dump('等待队列的处理'.count($res));
    72.                             //$this->server->send($fd,'123');
    73.                         });
    74.                   }
    75.              }
    76.         });*/
    79.     }
    81.     public  function  onReceive($server,$fd,$reactor_id,$data){
    82.         $len=unpack('N',$data)[1];
    83.         $body=substr($data,-$len);//去除二进制数据之后,不要包头的数据
    84.        // try{
    85.             //连接数不够了
    86.             if(count($this->idle_pool)!=0){
    87.                 $this->query($body,function ($res) use($fd){
    88.                     var_dump('最小连接的执行结果'.count($res));
    89.                     //$this->server->send($fd,'123');
    90.                 });
    92.             }else{
    93.                 //重新建立连接,并且判断当前的连接池数量小于最大的连接数
    95.                 //var_dump('连接池个数'.$this->count);
    97.                 $count=$this->table->get(1,'count');
    99.                 if($count<$this->max){
    100.                     $this->table->incr(1,'count');
    101.                     $this->db_connect($this->db_config,function() use($body){
    102.                         $this->query($body,function($res){
    103.                             var_dump('新创建的连接结果'.count($res));
    104.                         });
    106.                     });
    107.                 }else{
    108.                     if(count($this->wait_queue)<$this->wait_queue_max){
    109.                         $this->push_queue('wait_queue',$body);
    110.                     }
    111.                 }
    112.             }
    113.             //判断连接池当中,有没有空闲的数据库连接,如果没有空闲的连接,并且等待队列未满
    114.             /*//放入到等待的队列当中
    115.             if(count($this->idle_pool)===0){
    116.                 if(count($this->wait_queue)<$this->wait_queue_max){
    117.                     $this->push_queue('wait_queue',$body);
    118.                 }
    119.             }*/
    121.        // }catch(\Exception $e){
    122.            // echo '捕获异常'.$e->getMessage();
    123.        // }
    125.     }
    127.     protected  function query($sql,$callback=''){
    129.         //取出可用连接
    130.         $idle_pool=$this->idle_pool->shift();
    131.         $db=$idle_pool['db'];
    132.         $db->query($sql,function ($db,$res) use ($callback){
    133.                  $callback($res);
    134.                  //执行完毕,重新放回到空闲连接池,复用连接
    135.                  $this->push_queue('idle_pool',$db);
    136.                  //取出队列当中的
    137.                 if(count($this->wait_queue)>0){
    138.                    //空闲连接池当中去取出连接去执行sql
    139.                    $idle_num=count($this->idle_pool);
    140.                    for($i=0;$i<$idle_num;$i++){
    141.                     $req=$this->wait_queue->shift();
    142.                     $this->query($req['db'],function ($res){
    143.                         //var_dump($res);
    144.                          var_dump('等待队列的处理'.count($res));
    145.                         //$this->server->send($fd,'123');
    146.                     });
    147.                  }
    148.             }
    150.         });
    153.     }
    154.     //向队列当中添加连接
    155.     public  function  push_queue($queueName,$pdo,$fd=0){
    156.         $arr=[
    157.             'fd'=>$fd,
    158.             'db'=>$pdo
    159.         ];
    160.         $this->$queueName->push($arr);
    161.     }
    164.     /**
    165.      * 数据库长连接
    166.      * @param $db_config
    167.      * @return array|PDO
    168.      */
    169.     public  function  db_connect($db_config,$callback=''){
    170.         $mysql=new swoole\mysql;
    171.         $mysql->connect($db_config['db'],function ($db,$res) use($callback){
    172.             //错误提示
    173.             if ($res == false) {
    174.                 var_dump('错误信息'.$db->connect_errno,$db->connect_error);
    175.                 return;
    176.             }
    177.             //维护数据库连接数量
    178.             $this->table->incr(1,'count');
    179.             //放入到空闲的连接池当中
    180.             $this->push_queue('idle_pool',$db);
    181.             if(is_callable($callback)){
    182.                 $callback();
    183.                 return;
    184.             }
    186.         });
    187.     }
    189. }
    191. $config['tcp']=[
    192.     'worker_num'=>1,
    193.     'open_length_check' => true,
    194.     'package_length_type'=>'N',
    195.     'package_length_offset'=>0, //计算总长度
    196.     'package_body_offset'=>4,//包体位置
    197. ];
    199. $config['server']=[
    200.     'max'=>20,
    201.     'min'=>5
    202. ];
    204. $config['db']=[
    205.     'host' => '127.0.0.1',
    206.     'port' => 3306,
    207.     'user' => 'root',
    208.     'password' => 'Qq!990979940',
    209.     'database' => 'test',
    210.     'charset' => 'utf8', //指定字符集
    211.     'timeout' =>10,  // 可选：连接超时时间（非查询超时时间），默认为SW_MYSQL_CONNECT_TIMEOUT（1.0）
    212. ];
    213. new DBServer($config);