1. <?php
    2. /**
    3.  * Created by PhpStorm.
    4.  * User: Administrator
    5.  * Date: 2018/5/23 0023
    6.  * Time: 下午 8:53
    7.  */
    9. class Pool{
    11.     protected  $table;
    12.     protected  $mpid; //进程id
    14.     public  function  __construct()
    15.     {
    17.             $this->mpid=getmypid();
    18.             $this->table=$this->create_table(); //将进程pid跟任务内容进行绑定
    19.             $this->run();
    20.             $this->signal();//信号监听
    21.     }
    24.     //获取任务,创建进程
    25.     protected  function  run(){
    26.         $data=[
    27.             '990979940@qq.com',
    28.             '213123@qq.com',
    29.             '457567657@qq.com'
    30.         ];
    32.         foreach ($data as $v){
    33.             $process=$this->create_process();
    34.             //pid=>990979940@qq.com  pid=>任务内容
    35.              $this->table->set($process->pid,['data'=>$v]);
    36.              $process->write($v);
    37.         }
    39.        /*foreach($this->table as $row)
    40.         {
    41.             var_dump($row);
    42.         }*/
    43.     }
    45.     //创建一个内存表
    46.     protected  function  create_table(){
    47.         $table=new swoole\table(1024);
    48.         $table->column('attempts',swoole\table::TYPE_INT,2);
    49.         $table->column('data',swoole\table::TYPE_STRING,64);
    50.         $table->create();
    51.         return $table;
    52.     }
    55.     /**
    56.      * 创建子进程
    57.      */
    58.     protected  function  create_process(){
    59.         $process=new swoole\Process([$this,'callback_function']);
    60.         $process->start(); //启动子进程
    61.           //异步读取管道消息
    62.           swoole_event::add($process->pipe,function () use($process){
    63.                 $data=json_decode($process->read(),true);
    64.                 if($data['status']==false);
    65.                 swoole_event::del($process->pipe);
    66.                 echo '子进程发送的内容:'.PHP_EOL;
    67.                 var_dump($data);
    68.           });
    70.           return $process;
    71.     }
    73.     //子进程业务处理逻辑
    74.     public  function callback_function($worker){
    75.         $res=$worker->read();
    76.         $rand=mt_rand(1,2);
    78.         for ($i=0;$i<20;$i++){
    79.             echo '子进程在执行'.$i.PHP_EOL;
    80.             $this->checkPid($worker,$this->mpid);//检测主进程是否异常终止避免子进程成为僵尸进程
    81.             sleep(1);
    82.         }
    84.         if($rand==1){
    85.             throw  new Exception('出现异常了');
    86.         }
    88.        /*//执行发短信的逻辑,正常的逻辑
    89.         $res=false;
    90.         if($res==false){
    91.             $data=[
    92.                 'status'=>false,'data'=>$res,'msg'=>'任务执行成功'
    93.             ];
    94.             $worker->write(json_encode($data)); //写入到主进程
    95.             return;
    96.         }*/
    98.         $data=[
    99.             'status'=>true,'data'=>$res,'msg'=>'任务执行成功'
    100.         ];
    101.         $worker->write(json_encode($data)); //写入到主进程
    103.         $this->table->del($worker->pid); //删除关系绑定的记录
    105.     }
    107.     /**
    108.      *
    109.      *检测主进程是否存在
    110.      * @param $mpid 父进程pid
    111.      *
    112.      */
    113.     public  function  checkPid($worker,$mpid){
    115.          if(!swoole\Process::kill($mpid,0)){
    117.              //等待子进程任务处理完毕之后再去关闭,根据业务场景,如果需要跟主进程交互的,主进程关闭了
    118.              //子进程需要关闭,如果不需要的,等待子进程执行完毕之后再关闭
    120.              $msg="主进程已经退出,工作进程{$worker->pid}退出";
    121.              echo $msg.PHP_EOL;
    122.              $worker->exit();//子进程退出
    123.          }
    124.     }
    126.     //捕获子进程结束时的信号,回收子进程
    127.     public  function  signal(){
    129.         swoole_process::signal(SIGCHLD, function($sig) {
    130.             //必须为false，非阻塞模式
    131.             while($ret =  swoole_process::wait(false)) {
    132.                    if($ret['code']>0){ //出现非0状态码
    133.                         $pid=$ret['pid'] ;//子进程结束pid
    134.                         $data=$this->table->get($pid);
    135.                        $this->table->del($pid); //删除之前内容
    136.                        //判断重试次数
    137.                        if($data['attempts']==0){
    138.                             $process=$this->create_process(); //重新拉起进程,会产生新的pid
    139.                             $this->table->set($process->pid,['data'=>$data['data'],'attempts'=>$data['attempts']+1]); //新创建的进程
    140.                             $process->write($data['data']);
    141.                             return;
    142.                         }
    143.                         //file_put_contents(); 记录日志|发送邮件|异常队列当中
    144.                    }
    145.             }
    146.            /* foreach($this->table as $row) {
    147.                   var_dump($row);
    148.             }*/
    150.         });
    151.     }
    153. }
    155. new Pool();