Process->useQueue

启用消息队列作为进程间通信。

  1. bool swoole_process->useQueue(int $msgkey = 0, int $mode = 2);

useQueue方法接受2个可选参数。

  • $msgkey是消息队列的key，默认会使用ftok(FILE, 1)作为KEY
  • $mode通信模式，默认为2，表示争抢模式，所有创建的子进程都会从队列中取数据
  • 如果创建消息队列失败，会返回false。可使用swoole_strerror(swoole_errno()) 得到错误码和错误信息。
使用模式2后，创建的子进程无法进行单独通信，比如发给特定子进程。 $process对象并未执行start，也可以执行push/pop向队列推送/提取数据 消息队列通信方式与管道不可共用。消息队列不支持EventLoop，使用消息队列后只能使用同步阻塞模式
  1. CygWin 环境不支持消息队列，请勿在此环境下使用

非阻塞

在1.9.2或更高版本中增加了swoole_process::IPC_NOWAIT的支持，可将队列设置为非阻塞。在非阻塞模式下，队列已满调用push方法、队列已空调用pop方法时将不再阻塞立即返回。

  1. //设置为非阻塞模式
  2. $process->useQueue($key, $mode | swoole_process::IPC_NOWAIT);

示例

  1. <?php
  2. $worker_num = 2;
  3. $process_pool = [];
  5. $process= null;
  6. $pid = posix_getpid();
  8. function sub_process(swoole_process $worker)
  9. {
  10.     sleep(1); //防止父进程还未往消息队列中加入内容直接退出
  11.     echo "worker ".$worker->pid." started".PHP_EOL;
  12.     while($msg = $worker->pop()){
  13.         if ($msg === false) {
  14.             break;
  15.         }
  16.         $sub_pid = $worker->pid;
  17.         echo "[$sub_pid] msg : $msg".PHP_EOL;
  18.         sleep(1);//这里的sleep模拟任务耗时，否则可能1个worker就把所有信息全接受了
  19.     }
  20.     echo "worker ".$worker->pid." exit".PHP_EOL;
  21.     $worker->exit(0);
  22. }
  24. $customMsgKey = 1;
  25. $mod = 2 | swoole_process::IPC_NOWAIT;//这里设置消息队列为非阻塞模式
  27. //创建worker进程
  28. for($i=0;$i<$worker_num; $i++) {
  29.     $process=new swoole_process('sub_process');
  30.     $process->useQueue($customMsgKey, $mod);
  31.     $process->start();
  32.     $pid = $process->pid;
  33.     $process_pool[$pid] = $process;
  34. }
  36. $messages = [
  37.     "Hello World!",
  38.     "Hello Cat!",
  39.     "Hello King",
  40.     "Hello Leon",
  41.     "Hello Rose"
  42. ];
  43. //由于所有进程是共享使用一个消息队列，所以只需向一个子进程发送消息即可
  44. $process = current($process_pool);
  45. foreach ($messages as $msg) {
  46.     $process->push($msg);
  47. }
  49. swoole_process::wait();
  50. swoole_process::wait();
  52. echo "master exit".PHP_EOL;