PHP中的子进程和消息队列

发布于 2018-01-10  1.34k 次阅读


本文将介绍PHP子进程的使用,使用linux消息队列机制来达成进程间的协作,最后用一个简单的例子来类比具体应用方案。

1. 子进程

1.1 创建子进程

int pcntl_fork ( void )

 

按照php官方的说明,pcntl_fork()函数会创建一个子进程,这个子进程仅PID(进程号) 和PPID(父进程号)与其父进程不同。成功时,在父进程执行线程内返回产生的子进程的PID,在子进程执行线程内返回0。失败时,在父进程上下文返回-1,不会创建子进程,并且会引发一个PHP错误。

fork之后,操作系统会复制一个与父进程近似的子进程,这2个进程共享代码空间,但是数据空间是互相独立的,子进程数据空间中的内容是父进程的完整拷贝,指令指针也完全相同,只有对于fork调用的返回会在父子进程中产生区别(见上)。简单理解就是fork之后,会产生一个新的子进程执行之后的代码。

至于哪一个进程最先运行,这与操作系统平台的调度算法有关,而且这个问题在实际应用中并不重要,对于父子进程协同运作,我们将介绍消息队列的使用。

1.2 结束子进程

当子进程处理完自己的任务,需要直接结束时,我们直接关闭该进程即可:

int posix_getpid ( void )

使用该函数,返回进程 id 号,为整型(integer)。在子进程中,该函数返回的即是子进程的进程号。

bool posix_kill ( int $pid , int $sig)

传递sig信号给pid对应的进程,关闭这个进程。sig参数可以百度“PHP信号管理”扩展阅读,这里我们使用简单的SIGTERM为终止进程用的软件终止信号。而pid使用刚刚介绍的获取进程id的函数即可获取。

 

2. 消息队列

2.1 创建消息队列

int ftok ( string $pathname , string $proj)

将一个可访问的文件路径名转换为一个可供 shmop_open() (此函数用于生产共享内存空间,即我们的消息队列)和其他系统VIPC keys使用的整数,proj参数必须是一个字符串,这个参数其实就是读写方式,一般使用a即可。

使用该函数,我们就可以简单生成一个消息队列用的键值。

 resource msg_get_queue ( int $key [,int $perms = 0666 ] )

msg_get_queue()会根据传入的键值(即使用上一个函数获取的)返回一个消息队列的引用。如果linux系统中没有消息队列与键值对应,msg_get_queue()将会创建一个新的消息队列。

函数的第二个参数需要传入一个int值,作为新创建的消息队列的权限值,默认为0666。这个权限值与linux命令chmod中使用的数值是一致的。

该函数的返回值是一个用于访问消息队列的句柄。

 

2.2 发送消息

bool msg_send ( resource $queue , int $msgtype , mixed $message)

实际定义还有一些参数可以传,由于比较冷门就不介绍了。这里queue对应消息队列的句柄。msgtype涉及到接收消息时候的操作,如果msgtype=0,接收消息队列的第一个消息;大于0接收队列中消息类型等于这个值的第一个消息;小于0接收消息队列中小于或者等于msgtype绝对值的所有消息中的最小一个消息。最后的message是传递的消息。

 

2.3 接受消息

bool msg_receive ( resource $queue , int $desiredmsgtype ,int &$msgtype , int $ maxsize, mixed &$message [, bool $unserialize =true [, int $flags = 0 [, int &$errorcode ]]] )

同样是非常长的一个函数,参数涉及到:

queue对应消息队列的句柄。

desiredmsgtype定义和发送用的msytype类似,只是0表示消息队列中的首个消息会被接受。

msgtype接受到消息的类型

maxsize接受消息的最大长度,如果消息长度超过这个限制,那么获取消息就会失败。

message消息本身,使用引用的方式获取值。

unserialize为true的情况下,不会对收到的信息进行二值化,也就是可以收到其他php脚本发来的数组或是复杂结构;为false的情况下,返回的将是字符串。

Flags函数的处理方式。MSG_IPC_NOWAIT,直接返回,如果失败则返回一个MSG_ENOMSG的整数。 MSG_EXCEPT,使用后当desiredmsgtype大于0 时,可获取第一条不为desiredmsgtype的消息。MSG_NOERROR如果消息超过了最大长度,会进行截断并不返回错误。

 

3. 实例

介绍了这么多函数,其实根本就不知道该怎么用,这里设计了一个比较贴近生产环境的流程:

在父进程中,我们创建了5个子进程来一起处理任务,例子中是简单的echo,实际中可能是统计某日的数据,或者是计算用户答案是否正确。之后每个子进程会不断读取消息队列中的消息,在处理到一定数量的消息后,结束这个子进程。而父进程在创建完子进程后,只需要不断的在消息队列中写入需要处理的任务即可。

<?php
$message_queue_key= ftok(__FILE__, 'a');
$message_queue= msg_get_queue($message_queue_key, 0666);
 
$pids= array();
for( $i = 0; $i < 5; $i++) {
    $pids[$i] = pcntl_fork();
    if ($pids[$i]) {
        echo "No.$i child process wascreated, the pid is $pids[$i]\r\n";
    } elseif ($pids[$i] == 0) {
        $pid = posix_getpid();
        echo "process.$pidstart\r\n";
        $count = 0;
        do {
            msg_receive($message_queue, 0,$message_type, 1024, $message, true, MSG_IPC_NOWAIT);
            echo "process.$pid dealmessage{$message}\r\n";
            $count++;
            if($count == 5) {
                break;
            }
            sleep(1);
        } while (true);
        echo "process.$pid end\r\n";
        posix_kill($pid, SIGTERM);
    }
}
for( $i = 0; $ i< 25; $i++) {
    msg_send($message_queue, 1,rand(1000,10000));
}
?>

 

处理的log如下,由于例子的输出有点长,这里选取主要的部分。可以看到子进程会如期接收到父进程写的任务,并且各自同时完成执行。

No.0 child process was created, thepid is 26512
process.26512 start
process.26512 deal message2513
No.1 child process was created, thepid is 26513
……
process.26513 deal message4692
process.26512 deal message5297
process.26512 end
……

 

转载自:https://blog.csdn.net/baidu_zhongce/article/details/49151689


公交车司机终于在众人的指责中将座位让给了老太太