摘要：消息隊(duì)列更常見的用途是主進(jìn)程分配任務(wù)，子進(jìn)程消費(fèi)執(zhí)行。子進(jìn)程前面加了個(gè)，這是為了防止父進(jìn)程還未往消息隊(duì)列中加入內(nèi)容直接退出。

前面幾節(jié)都是講解pcntl擴(kuò)展實(shí)現(xiàn)的多進(jìn)程程序。本節(jié)給大家介紹swoole擴(kuò)展的swoole_process模塊。

swoole_process 是swoole提供的進(jìn)程管理模塊，用來替代PHP的pcntl擴(kuò)展。

首先，確保安裝的swoole版本大于1.7.2：

$ php --ri swoole

swoole

swoole support => enabled
Version => 1.10.1

注意：swoole_process在最新的1.8.0版本已經(jīng)禁止在Web環(huán)境中使用了，所以也只能支持命令行。

swoole提供的多進(jìn)程擴(kuò)展基本功能和pcntl提供的一樣，但swoole更易簡(jiǎn)單上手，并且提供了：

默認(rèn)基于unixsock的進(jìn)程間通信；

支持消息隊(duì)列作為進(jìn)程間通信；

基于signalfd和eventloop處理信號(hào)，幾乎沒有任何額外消耗；

高精度微秒定時(shí)器；

配合swoole_event模塊，創(chuàng)建的PHP子進(jìn)程可以異步的事件驅(qū)動(dòng)模式

swoole_process 模塊提供的方法(Method)主要分為四部分：

基礎(chǔ)方法

swoole_process::__construct
swoole_process->start
swoole_process->name
swoole_process->exec
swoole_process->close
swoole_process->exit
swoole_process::kill
swoole_process::wait
swoole_process::daemon
swoole_process::setAffinity

管道通信

swoole_process->write
swoole_process->read
swoole_process->setTimeout
swoole_process->setBlocking

消息隊(duì)列通信

swoole_process->useQueue
swoole_process->statQueue
swoole_process->freeQueue
swoole_process->push
swoole_process->pop

信號(hào)與定時(shí)器

swoole_process::signal
swoole_process::alarm

本例實(shí)現(xiàn)的是tcp server，特性：

多進(jìn)程處理客戶端連接

子進(jìn)程退出，Master進(jìn)程會(huì)重新創(chuàng)建一個(gè)

支持事件回調(diào)

主進(jìn)程退出，子進(jìn)程在干完手頭活后退出

mpid = $id = getmypid();   
            echo time()." Master process, pid {$id}
"; 

            //創(chuàng)建tcp server
            $this->socket = stream_socket_server("tcp://0.0.0.0:9201", $errno, $errstr);
            if(!$this->socket) exit("start server err: $errstr --- $errno");

            for($i=0; $istart_worker_process();
            }
    
            echo "waiting client...
";
    
            //Master進(jìn)程等待子進(jìn)程退出，必須是死循環(huán)
            while(1){
                foreach($this->pids as $k=>$pid){
                    if($pid){
                        $res = swoole_process::wait(false);
                        if ( $res ){
                            echo time()." Worker process $pid exit, will start new... 
";
                            $this->start_worker_process();
                            unset($this->pids[$k]);
                        }
                    }
                }
                sleep(1);//讓出1s時(shí)間給CPU
            }
        }, false, false); //不啟用管道通信
        swoole_process::daemon(); //守護(hù)進(jìn)程
        $process->start();//注意：start之后的變量子進(jìn)程里面是獲取不到的
    }

    /**
     * 創(chuàng)建worker進(jìn)程，接受客戶端連接
     */
    private function start_worker_process(){
        $process = new swoole_process(function(swoole_process $worker){
            $this->acceptClient($worker);
        }, false, false);
        $pid = $process->start();
        $this->pids[] = $pid;
    }

    private function acceptClient(&$worker)
    {
        //子進(jìn)程一直等待客戶端連接，不能退出
        while(1){
            
            $conn = stream_socket_accept($this->socket, -1);
            if($this->onConnect) call_user_func($this->onConnect, $conn); //回調(diào)連接事件

            //開始循環(huán)讀取消息
            $recv = ""; //實(shí)際收到消息
            $buffer = ""; //緩沖消息
            while(1){
                $this->checkMpid($worker);
                
                $buffer = fread($conn, 20);

                //沒有收到正常消息
                if($buffer === false || $buffer === ""){
                    if($this->onClose) call_user_func($this->onClose, $conn); //回調(diào)斷開連接事件
                    break;//結(jié)束讀取消息，等待下一個(gè)客戶端連接
                }

                $pos = strpos($buffer, "
"); //消息結(jié)束符
                if($pos === false){
                    $recv .= $buffer;                            
                }else{
                    $recv .= trim(substr($buffer, 0, $pos+1));

                    if($this->onMessage) call_user_func($this->onMessage, $conn, $recv); //回調(diào)收到消息事件

                    //客戶端強(qiáng)制關(guān)閉連接
                    if($recv == "quit"){
                        echo "client close conn
";
                        fclose($conn);
                        break;
                    }

                    $recv = ""; //清空消息，準(zhǔn)備下一次接收
                }
            }
        }
    }

    //檢查主進(jìn)程是否存在，若不存在子進(jìn)程在干完手頭活后退出
    public function checkMpid(&$worker){
        if(!swoole_process::kill($this->mpid,0)){
            $worker->exit();
            // 這句提示,實(shí)際是看不到的.需要寫到日志中
            echo "Master process exited, I [{$worker["pid"]}] also quit
";
        }
    }

    function __destruct() {
        @fclose($this->socket);
    }
}

$server =  new TcpServer();

$server->onConnect = function($conn){
    echo "onConnect -- accepted " . stream_socket_get_name($conn,true) . "
";
    fwrite($conn,"conn success
");
};

$server->onMessage = function($conn,$msg){
    echo "onMessage --" . $msg . "
";
    fwrite($conn,"received ".$msg."
");
};

$server->onClose = function($conn){
    echo "onClose --" . stream_socket_get_name($conn,true) . "
";
    fwrite($conn,"onClose "."
");
};

$server->run();

運(yùn)行后可以使用telnet連接：

telnet 127.0.0.1 9201

由于設(shè)置了最大三個(gè)子進(jìn)程，最多只能接受3個(gè)客戶端連接。

前面講解的例子里，主進(jìn)程和子進(jìn)程直接是沒有直接的數(shù)據(jù)交互的。如果主進(jìn)程需要得到的來自子進(jìn)程的反饋，或者子進(jìn)程接受來自主進(jìn)程的數(shù)據(jù)，那么就需要進(jìn)程間通信了。

swoole內(nèi)置了管道通信和消息隊(duì)列通信。

管道通信主要是數(shù)據(jù)傳輸：一個(gè)進(jìn)程需要將數(shù)據(jù)發(fā)送給另外一個(gè)進(jìn)程。

這個(gè)swoole封裝后，使用非常簡(jiǎn)單：

read();

        ob_start();
        passthru($cmd);//執(zhí)行外部程序并且顯示未經(jīng)處理的、原始輸出，會(huì)直接打印輸出。
        $return = ob_get_clean() ? : " ";
        $return = trim($return).". worker pid:".$worker->pid."
";
        
        // $worker->write($return);//寫入數(shù)據(jù)到管道
        echo $return;//寫入數(shù)據(jù)到管道。注意：子進(jìn)程里echo也是寫入到管道
    }, true); //第二個(gè)參數(shù)為true，啟用管道通信
    $pid = $process->start();
    $workers[$pid] = $process;  
}

foreach($workers as $pid=>$worker){
    $worker->write("whoami"); //通過管道發(fā)數(shù)據(jù)到子進(jìn)程。管道是單向的：發(fā)出的數(shù)據(jù)必須由另一端讀取。不能讀取自己發(fā)出去的
    $recv = $worker->read();//同步阻塞讀取管道數(shù)據(jù)
    echo "recv result: $recv";
}

//回收子進(jìn)程
while(count($workers)){
    // echo time(). "
";
    foreach($workers as $pid=>$worker){
        $ret = swoole_process::wait(false);
        if($ret){
            echo "worker exit: $pid
";
            unset($workers[$pid]);
        }
    }
}

運(yùn)行：

$ php swoole_process_pipe.php 
recv result: Linux
recv result: 2018年 06月 24日 星期日 16:18:01 CST
recv result: yjc
worker exit: 14519
worker exit: 14522
worker exit: 14525

注意點(diǎn)：
1、管道數(shù)據(jù)讀取是同步阻塞的；上面的例子里如果子進(jìn)程里再加一句$worker->read()，會(huì)一直阻塞?？梢允褂?b>swoole_event_add將管道加入到事件循環(huán)中，變?yōu)楫惒侥Ｊ健?
2、子進(jìn)程里的輸出(例如echo)與write效果相同。
3、通過管道發(fā)數(shù)據(jù)到子進(jìn)程。管道是單向的：發(fā)出的數(shù)據(jù)必須由另一端讀取。不能讀取自己發(fā)出去的。

這里額外講解一下swoole_process::wait()：
1、swoole_process::wait()默認(rèn)是阻塞的, swoole_process::wait(false)則是非阻塞的；
2、swoole_process::wait()阻塞模式調(diào)用一次僅能回收一個(gè)子進(jìn)程，非阻塞模式調(diào)用一次不一定能當(dāng)前就能回收子進(jìn)程；
3、如果不加swoole_process::wait()，主進(jìn)程又是死循環(huán)，主進(jìn)程退出后會(huì)變成僵尸進(jìn)程。

ps -A -o stat,ppid,pid,cmd | grep -e "^[Zz]"可以查詢僵尸進(jìn)程。

防盜版聲明：本文系原創(chuàng)文章，發(fā)布于公眾號(hào)飛鴻影的博客(fhyblog)及博客園，轉(zhuǎn)載需作者同意。

消息隊(duì)列與管道有些不一樣：消息隊(duì)列是全局的，所有進(jìn)程都可以發(fā)送、讀取。你可以把它看做redis list結(jié)構(gòu)。

消息隊(duì)列更常見的用途是主進(jìn)程分配任務(wù)，子進(jìn)程消費(fèi)執(zhí)行。

pop()){
            // echo "recv from master: $cmd
";

            ob_start();
            passthru($cmd);//執(zhí)行外部程序并且顯示未經(jīng)處理的、原始輸出，會(huì)直接打印輸出。
            $return = ob_get_clean() ? : " ";
            $return = "res: ".trim($return).". worker pid: ".$worker->pid."
";
            
            echo $return;
            // sleep(1);
        }

        $worker->exit(0);
    }, false, false); //不創(chuàng)建管道

    $process->useQueue(1, 2 | swoole_process::IPC_NOWAIT); //使用消息隊(duì)列
    $pid = $process->start();
    $workers[$pid] = $process;
}

//由于所有進(jìn)程是共享使用一個(gè)消息隊(duì)列，所以只需向一個(gè)子進(jìn)程發(fā)送消息即可
$worker = current($workers);
for ($i=0; $i<3; $i++) {
    $worker->push("whoami"); //發(fā)送消息
}


//回收子進(jìn)程
while(count($workers)){
    foreach($workers as $pid=>$worker){
        $ret = swoole_process::wait();
        if($ret){
            echo "worker exit: $pid
";
            unset($workers[$pid]);
        }
    }
}

運(yùn)行結(jié)果：

$ php swoole_process_quene.php 
res: yjc. worker pid: 15885
res: yjc. worker pid: 15886
res: yjc. worker pid: 15887
worker exit: 15885
worker exit: 15886
worker exit: 15887

注意點(diǎn)：
1、所有進(jìn)程共享使用一個(gè)消息隊(duì)列；
2、消息隊(duì)列的讀取操作是阻塞的，可以在useQueue的時(shí)候第2個(gè)參數(shù)mode改為2 | swoole_process::IPC_NOWAIT，則是異步的。mode僅僅設(shè)置為2是阻塞的，示例里去掉swoole_process::IPC_NOWAIT后讀取消息的while會(huì)死循環(huán)。
3、子進(jìn)程前面加了個(gè)sleep(1);，這是為了防止父進(jìn)程還未往消息隊(duì)列中加入內(nèi)容直接退出。
4、子進(jìn)程末尾也加了sleep，這是為了防止一個(gè)進(jìn)程把所有消息都消費(fèi)完了，實(shí)際應(yīng)用需要去掉。

swoole_process::alarm支持微秒定時(shí)器：

 5) {
        //清除定時(shí)器
        swoole_process::alarm(-1);

        //退出進(jìn)程
        swoole_process::kill(getmypid());
        
    }
}

//安裝信號(hào)
swoole_process::signal(SIGALRM, "ev_timer");

//觸發(fā)定時(shí)器信號(hào)：?jiǎn)挝粸槲⒚?。如果為?fù)數(shù)表示清除定時(shí)器
swoole_process::alarm(100 * 1000);//100ms

echo getmypid()."
"; //該句會(huì)順序執(zhí)行，后續(xù)無需使用while循環(huán)防止進(jìn)程直接退出

運(yùn)行：

$ php swoole_process_alarm.php 
13660
#0    alarm
#1    alarm
#2    alarm
#3    alarm
#4    alarm
#5    alarm

注：alarm不能和SwooleTimer同時(shí)使用。

1、Process-Swoole-Swoole文檔中心
https://wiki.swoole.com/wiki/...

歡迎關(guān)注公眾號(hào)及時(shí)獲取最新文章推送！

推薦！每月僅需$2.5，即可擁有配置SSD的VPS！