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秒，不然主进程pop时 获取消息太快，而阻塞状态了，push完给子进程消息，子进程逻辑还没处理完            //(1)不加这停两秒时，可能主进程马上pop时，队列是空的，主进程阻塞状态了，下面子进程只能pop和push一次，再pop时也阻塞状态了            //客户端再发send消息时,服务器不响应，除非新client进来，发send，主进程阻塞取消了，第一个客户端又可以发send消息了，第二个客户端            //发send消息时，服务端不响应，这种情况发生了            $result = $process->pop();//默认模式下，如果队列中没有数据，pop方法会阻塞等待            echo "From worker: $result\n";//这里主进程，接受到的子进程的数据        }

        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");//子进程处理完逻辑，不是马上push数据给主进程  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";