swoole process use queue example
更新:HHH   时间:2023-1-7


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";


返回web开发教程...