PHP多进程通信-消息队列使用

239 阅读 0 评论 158 点赞

我是靠谱客的博主独特犀牛，这篇文章主要介绍PHP多进程通信-消息队列使用，现在分享给大家，希望可以做个参考。

向消息队列发送数据和获取数据的测试

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
<?php
$key=ftok(__FILE__,'a');

//获取消息队列
$queue=msg_get_queue($key,0666);
//发送消息
//msg_send($queue, 1, "Hello, 1");
//接收消息，如果接收不到会阻塞
msg_receive($queue, 1, $message_type, 1024, $message1);
//移除消息
//msg_remove_queue($queue);
//var_dump($message1);

复制代码

<?php
/**
 * 这段代码模拟了一个日常的任务。
 * 第一个父进程产生了一个子进程。子进程又作为父进程，产生10个子进程。
 * 可以简化为A -> B -> c,d,e... 等进程。
 * 作为A来说，只需要生产任务，然后交给B 来处理。B 则会将任务分配给10个子进程来进行处理。
 * 
 */

//设定脚本永不超时
set_time_limit(0);
$ftok = ftok(__FILE__, 'a');
$msg_queue = msg_get_queue($ftok);
$pidarr = [];

//产生子进程
$pid = pcntl_fork();
if ($pid) {
 //父进程模拟生成一个特大的数组。
 $arr = range(1,100000);
 //将任务放进队里，让多个子进程并行处理
 foreach ($arr as $val) {
  $status = msg_send($msg_queue,1, $val);
  usleep(1000);
 } 
 $pidarr[] = $pid;
 msg_remove_queue($msg_queue);
} else {
 //子进程收到任务后，fork10个子进程来处理任务。
 for ($i =0; $i<10; $i++) {
  $childpid = pcntl_fork();
  if ($childpid) {
   $pidarr[] = $childpid; //收集子进程processid
  } else {
   while (true) {
    msg_receive($msg_queue, 0, $msg_type, 1024, $message);
    if (!$message) exit(0);
    echo $message.PHP_EOL;
    usleep(1000);
   } 
  } 
 } 
}

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
<?php
/**
 * 这段代码模拟了一个日常的任务。
 * 第一个父进程产生了一个子进程。子进程又作为父进程，产生10个子进程。
 * 可以简化为A -> B -> c,d,e... 等进程。
 * 作为A来说，只需要生产任务，然后交给B 来处理。B 则会将任务分配给10个子进程来进行处理。
 * 
 */

//设定脚本永不超时
set_time_limit(0);
$ftok = ftok(__FILE__, 'a');
$msg_queue = msg_get_queue($ftok);
$pidarr = []; 

//产生子进程
$pid = pcntl_fork();
if ($pid) {
 //父进程模拟生成一个特大的数组。
 $arr = range(1,100000);
 //将任务放进队里，让多个子进程并行处理
 foreach ($arr as $val) {
  $status = msg_send($msg_queue,1, $val);
  usleep(1000);
 } 
 $pidarr[] = $pid;
 msg_remove_queue($msg_queue);
} else {
 //子进程收到任务后，fork10个子进程来处理任务。
 for ($i =0; $i<10; $i++) {
  $childpid = pcntl_fork();
  if ($childpid) {
   $pidarr[] = $childpid; //收集子进程processid
  } else {
   while (true) {
    msg_receive($msg_queue, 0, $msg_type, 1024, $message);
    if (!$message) exit(0);
    echo $message.PHP_EOL;
    usleep(1000);
   } 
  } 
 } 
}

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