php消息队列实现详解

<?php
$key = ftok('demo1.php',"x");//将文件与ID转换为一个key，这个文件只要真实存在就行，没有特殊限制，一般都指定当前文件名
$msqid = msg_get_queue($key);// 创建一个消息队列，返回一个资源id
echo msg_send($msqid,1,"hello"); //往消息队列写入一条数据 hello 到队列中
echo $msqid; // 输出资源id

<?php
$a = serialize("hello");
echo 'hello 序列化长度：'.strlen($a)."\n";
echo 'hello 序列化内容：'.$a."\n";
echo "hello 未序列化长度：".strlen("hello")."\n";

<?php
$key = ftok('demo1.php',"x");//将文件与ID转换为一个key
$msqid = msg_get_queue($key);//消息队列如果已创建，直接返回同一个资源描述符
//第一个参数资源描述符，第二个参数一般为0，表示从队列第一条读取，或者设置与msg_send函数第二个参数一致，第三个参数设置缓冲区长度，注意msg_send 函数默认会序列化数据，msg_receive函数默认会将读取数据反序列化，如果缓冲区设置过小会反序列化失败
msg_receive($msqid,1,$received_message_type,1024,$message);//读取消息队列数据，逐条读取，
echo $message.PHP_EOL;

<?php
$key = ftok('demo3.php',"x");//将文件与ID转换为一个key，这个文件只要真实存在就行，没有特殊限制，一般都指定当前文件名
$msqid = msg_get_queue($key);// 创建一个消息队列，返回一个资源id
// 第一个参数资源id,第二个参数消息类型，必须大于0，第三个参数写入队列内容，取消序列化操作
echo msg_send($msqid,1,"hello",false); //往消息队列写入一条数据 hello 到队列中

<?php
$key = ftok('demo3.php',"x");//将文件与ID转换为一个key
$msqid = msg_get_queue($key);
//第一个参数资源描述符
//第二个参数一般为0，表示从队列第一条读取，或者设置与msg_send函数第二个参数一致
// 第三个参数是msg_send函数的第二个参数的值，也就是发送消息的消息类型
//第四个参数设置接收缓冲区长度，如果设置过小会反序列化失败【如果使用序列化功能时候】
//当接收缓冲区大小设置过小并且关闭了反序列化【发送端发送数据也关闭了】，但是发送的数据超过了设置的长度，就会被截断，并且额外的数据会被丢弃，必须设置第五个参数才有效
// 第五个参数 flags 标志符
msg_receive($msqid,0,$received_message_type,1,$message,false,MSG_NOERROR);
echo $message.PHP_EOL;

<?php
$key = ftok('demo5.php',"x");//将文件与ID转换为一个key
$msqid = msg_get_queue($key);
// fork 一个子进程
$pid = pcntl_fork();
if($pid == 0){
    while(1){
    // MSG_IPC_NOWAIT【启用非阻塞】 msg_receive 函数其实是封装 linux c msgrcv 系统调用函数，如果设置为非阻塞，函数调用次数非常高，会非常消耗cpu资源
    //阻塞模式调用函数次数低，有数据才返回，对cpu友好
        $ret = msg_receive($msqid,0,$msgType,1024,$msg,true,MSG_IPC_NOWAIT,$error_code);
        if($error_code != MSG_ENOMSG){
            echo $msg."\n";
        }
        sleep(2);
        echo "go"."\n";
    }
    exit(0);
}

<?php
$key = ftok('demo5.php',"x");//将文件与ID转换为一个key
$msqid = msg_get_queue($key);
$pid = pcntl_fork();
if($pid == 0){
    while(1){
        $ret = msg_receive($msqid,0,$msgType,1024,$msg,true,MSG_IPC_NOWAIT,$error_code);
        if($error_code != MSG_ENOMSG){
            echo $msg."\n";
        }
    }
    exit(0);
}
$i = 1;
while(1){
    msg_send($msqid,2,"hello world",true);
    sleep(2);
    if($i++ == 3){
    	// 杀死子进程
        posix_kill($pid,SIGILL);
        break;
    }
}
//子进程退出回收子进程，防止孤儿进程
$pid = pcntl_wait($status);
if($pid > 0)
{
    echo "exit pid=".$pid."\n";
}

<?php
$key = ftok('demo5.php',"x");//将文件与ID转换为一个key
$msqid = msg_get_queue($key);
$pid = pcntl_fork();
if($pid == 0){
    while(1){
        $ret = msg_receive($msqid,0,$msgType,1024,$msg,true,MSG_IPC_NOWAIT,$error_code);
        if($error_code != MSG_ENOMSG){
            echo $msg."\n";
        }

    }
    exit(0);
}
$i = 1;
while(1){
    msg_send($msqid,2,"hello world",true);
    sleep(2);
    if($i++ == 3){
    	// 杀死子进程
        posix_kill($pid,SIGILL);
        break;
    }
}
//子进程退出回收子进程，防止孤儿进程
$pid = pcntl_wait($status);
if($pid > 0)
{
    echo "exit pid=".$pid."\n";
}
//msg_remove_queue 函数移除消息队列
if(msg_remove_queue($msqid)){
    echo "remove ok\n";
}