PHP多進(jìn)程通信-消息隊列使用

?php
$key=ftok(__FILE__,'a');

//獲取消息隊列
$queue=msg_get_queue($key,0666);
//發(fā)送消息
//msg_send($queue, 1, "Hello, 1");
//接收消息，如果接收不到會阻塞
msg_receive($queue, 1, $message_type, 1024, $message1);
//移除消息
//msg_remove_queue($queue);
//var_dump($message1);

?php
/**
 * 這段代碼模擬了一個日常的任務(wù)。
 * 第一個父進(jìn)程產(chǎn)生了一個子進(jìn)程。子進(jìn)程又作為父進(jìn)程，產(chǎn)生10個子進(jìn)程。
 * 可以簡化為A -> B -> c,d,e... 等進(jìn)程。
 * 作為A來說，只需要生產(chǎn)任務(wù)，然后交給B 來處理。B 則會將任務(wù)分配給10個子進(jìn)程來進(jìn)行處理。
 * 
 */

//設(shè)定腳本永不超時
set_time_limit(0);
$ftok = ftok(__FILE__, 'a');
$msg_queue = msg_get_queue($ftok);
$pidarr = []; 

//產(chǎn)生子進(jìn)程
$pid = pcntl_fork();
if ($pid) {
 //父進(jìn)程模擬生成一個特大的數(shù)組。
 $arr = range(1,100000);
 //將任務(wù)放進(jìn)隊里，讓多個子進(jìn)程并行處理
 foreach ($arr as $val) {
  $status = msg_send($msg_queue,1, $val);
  usleep(1000);
 } 
 $pidarr[] = $pid;
 msg_remove_queue($msg_queue);
} else {
 //子進(jìn)程收到任務(wù)后，fork10個子進(jìn)程來處理任務(wù)。
 for ($i =0; $i10; $i++) {
  $childpid = pcntl_fork();
  if ($childpid) {
   $pidarr[] = $childpid; //收集子進(jìn)程processid
  } else {
   while (true) {
    msg_receive($msg_queue, 0, $msg_type, 1024, $message);
    if (!$message) exit(0);
    echo $message.PHP_EOL;
    usleep(1000);
   } 
  } 
 } 
}

//防止主進(jìn)程先于子進(jìn)程退出，形成僵尸進(jìn)程
while (count($pidarr) > 0) {
 foreach ($pidarr as $key => $pid) {
  $status = pcntl_waitpid($pid, $status);
  if ($status == -1 || $status > 0) {
   unset($pidarr[$key]);
  } 
 } 
 sleep(1);
}