swoole process use queue example

server:

class Server
{
    private $_serv = null;

    private $_workers = [];

    private $_worker_num = 2;

    public function __construct()
    {
        $this->_serv = new swoole_server('127.0.0.1', 9501);

        $this->_serv->on('start', array($this, 'onStart'));
        $this->_serv->on('workerstart', array($this, 'onWorkerStart'));
        $this->_serv->on('connect', array($this, 'onConnect'));
        $this->_serv->on('receive', array($this, 'onReceive'));
        $this->_serv->on('close', array($this, 'onClose'));

        $this->_serv->start();
    }

    public function onStart($serv)
    {
        echo "start \n";
    }

    public function onWorkerStart($serv, $worker_id)
    {
        echo "worker start\n";
    }
    
    public function onConnect($serv, $fd, $from_id )
    {
        echo "connect..\n";
    }

    public function onReceive(swoole_server $serv, $fd, $from_id, $data)
    {

        for ($i = 0; $i < $this->_worker_num ;$i++) {
            $process = new swoole_process(array($this, 'onProcess'), false, false);
            $process->useQueue();
            $pid = $process->start();
            echo $pid;
            $this->_workers[$pid] = $process;
        }

        foreach ($this->_workers as $pid => $worker) {
            echo $process->push("hello worker[{$pid}]\n");
            sleep(2);//停2秒，不然主進(jìn)程pop時(shí) 獲取消息太快，而阻塞狀態(tài)了，push完給子進(jìn)程消息，子進(jìn)程邏輯還沒處理完
            //(1)不加這停兩秒時(shí)，可能主進(jìn)程馬上pop時(shí)，隊(duì)列是空的，主進(jìn)程阻塞狀態(tài)了，下面子進(jìn)程只能pop和push一次，再pop時(shí)也阻塞狀態(tài)了
            //客戶端再發(fā)send消息時(shí),服務(wù)器不響應(yīng)，除非新client進(jìn)來，發(fā)send，主進(jìn)程阻塞取消了，第一個(gè)客戶端又可以發(fā)send消息了，第二個(gè)客戶端
            //發(fā)send消息時(shí)，服務(wù)端不響應(yīng)，這種情況發(fā)生了
            
$result = $process->pop();//默認(rèn)模式下，如果隊(duì)列中沒有數(shù)據(jù)，pop方法會阻塞等待
            echo "From worker: $result\n";//這里主進(jìn)程，接受到的子進(jìn)程的數(shù)據(jù)
        }

        for($i = 0; $i < $this->_worker_num; $i++)
        {
            $ret = swoole_process::wait();
            $pid = $ret['pid'];
            unset($this->_workers[$pid]);
            echo "Worker Exit, PID=".$pid.PHP_EOL;
        }

    }

    public function onProcess($worker)
    {
        $msg_status = $worker->statQueue();
        if($msg_status['queue_num'] > 0)
        {
            $recv = $worker->pop();

            echo "FROM master {$recv}\n";
            $worker->push("heheh parent");//子進(jìn)程處理完邏輯，不是馬上push數(shù)據(jù)給主進(jìn)程  sleep(2)
        }
        $worker->exit(0);

    }

    public function onClose($serv, $fd, $from_id)
    {
        echo "close.\n";
    }
}

new Server();

client:

$cli = new Swoole_client(SWOOLE_SOCK_TCP);

$cli->connect('127.0.0.1', 9501, 1);

fwrite(STDOUT, '輸入消息：');
$msg = trim(fgets(STDIN));

echo $cli->send($msg);

echo $cli->recv()."\n";