1. <?php
    3. class pool{
    5.     protected  $server;
    6.     protected  $db_config; //配置信息
    7.     public  function  __construct($config)
    8.     {
    9.         //全局对象
    10.         $this->server=new \swoole\server('0.0.0.0',9800);
    11.         $this->server->set($config['tcp']);
    12.         //注册事件
    13.         $this->server->on('WorkerStart',[$this,'onWorkerStart']);
    14.         $this->server->on('receive',[$this,'onReceive']);
    15.         $this->server->on('task',[$this,'onTask']);
    16.         $this->server->on('finish',[$this,'onFinish']);
    17.         $this->db_config=$config;
    18.         //echo $this->idle_pool->shift();
    19.         $this->server->start(); //全局生命周期
    20.     }
    22.     public  function  onWorkerStart($server,$worker_id){
    23.     }
    25.     public  function  onReceive($server,$fd,$reactor_id,$data){
    26.         $len=unpack('N',$data)[1];
    27.         $body=substr($data,-$len);//去除二进制数据之后,不要包头的数据
    29.         //sql发送过来的时候,发送给连接池 task
    30.         //同步阻塞
    31.         $result=$server->taskwait($body);
    32.         var_dump($result);
    33.     }
    35.          /*
    36.      * 优点
    37.      *  1.简单实现
    38.      *  2.自带排队效果
    39.      *
    40.      *缺点
    41.      * 1.没有办法去控制连接池数量,资源非常昂贵是多进程形式的
    42.      * 2.多了一次进程间通讯
    43.      * 3.长连接复用连接
    44.      */
    45.     public  function  onTask($server,$task_id,$worker_id,$data){
    46.         $res=$this->query($data); //执行一条sql
    47.         $server->finish(json_encode($res));
    48.     }
    50.     public  function  onFinish($server,$task_id,$data){
    53.     }
    54.     protected  function query($sql){
    56.         //执行sql的时候,两次循环,如果执行成功,就跳出循环失败
    57.         $db=$this->db_connect($this->db_config);
    58.         for ($i=0;$i<2;$i++){
    59.             try{
    60.                 $res=$db->query($sql);
    61.                 break;
    62.             }catch (PDOException $e){
    63.                 $err_msg=$e->getMessage();
    64.                 if(strpos($err_msg,'MySQL server has gone away')!==false){
    65.                     $db=$this->db_connect($this->db_config);
    66.                 }
    67.                 $res=$db->query($sql);
    68.                 break;
    69.             }
    70.         }
    71.         //如果当前是select
    72.         if(preg_match('/^select/i',$sql)){
    73.             $data=$res->fetchAll(PDO::FETCH_ASSOC); //执行查询
    74.         }else{
    75.             $data=$res;
    76.         }
    77.         return $data;
    78.     }
    80.     /**
    81.      * 数据库长连接
    82.      * @param $db_config
    83.      * @return array|PDO
    84.      */
    85.     public  function  db_connect($db_config){
    86.         try{
    87.             $pdo=new PDO($db_config['db']['dsn'],$db_config['db']['user'],$db_config['db']['password'],[PDO::ATTR_PERSISTENT=>true,PDO::ATTR_SERVER_INFO=>true]);
    88.             return $pdo;
    89.         }catch (Exception $e){
    90.             echo $e->getMessage();
    91.             return [];
    92.         }
    93.     }
    95. }
    97. $config['tcp']=[
    98.     'worker_num'=>10,
    99.     'task_worker_num'=>20, //连接池数量
    100.     'open_length_check' => true,
    101.     'package_length_type'=>'N',
    102.     'package_length_offset'=>0, //计算总长度
    103.     'package_body_offset'=>4,//包体位置
    104. ];
    105. $config['server']=[
    106.     'max'=>20,
    107.     'min'=>2
    108. ];
    110. $config['db']=[
    111.     'dsn'=>'mysql:host=localhost;dbname=test',
    112.     'user'=>'root',
    113.     'password'=>'Qq!990979940'
    114. ];
    116. new Pool($config);