swooleprocessusequeueexample

server:

成都创新互联公司-专业网站定制、快速模板网站建设、高性价比八公山网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式八公山网站制作公司更省心,省钱,快速模板网站建设找我们，业务覆盖八公山地区。费用合理售后完善，十余年实体公司更值得信赖。

classServer
{
    private$_serv=null;

    private$_workers= [];

    private$_worker_num=2;

    public function__construct()
    {
        $this->_serv=newswoole_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 functiononStart($serv)
    {
        echo"start\";
    }

    public functiononWorkerStart($serv,$worker_id)
    {
        echo"worker start\";
    }
    
    public functiononConnect($serv,$fd,$from_id)
    {
        echo"connect..\";
    }

    public functiononReceive(swoole_server$serv,$fd,$from_id,$data)
    {

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

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

        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 functiononProcess($worker)
    {
        $msg_status=$worker->statQueue();
        if($msg_status['queue_num'] >0)
        {
            $recv=$worker->pop();

            echo"FROM master{$recv}\";
            $worker->push("heheh parent");//子进程处理完逻辑，不是马上push数据给主进程  sleep(2)
        }
        $worker->exit(0);

    }

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

newServer();

client:

$cli=newSwoole_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()."\";