php进程间通信消息队列

为什么需要进程

进程是系统进行资源分配和调度的基本单位。进程作为程序独立运行的载体，保障程序正常执行。进程的存在使得操作系统资源的利用率大幅提升。

消息队列

消息队列是在消息的传输过程中保存消息的容器。消息队列管理器相当于消息发送者和接收者的中介。消息队列的主要目的是创建路由并且保证消息可靠传递；如果发送消息时接收者不可用，消息队列会保留消息，直到有人接收它。

消息队列可提供临时存储的功能并且能保证消息可靠的传递，我们正好使用它实现进程间通信。当然消息队列不单单用于进程间通信，他的应用领域非常广。比如消息队列非常适用于解决消费者和生产者的问题，因为生产者和消费者之间总会存在『速度差』。比如生产者突然少了 10 个，两边处理的速度就会不平衡，会导致排队阻塞，服务不可用。这肯定不是我们想看到的，如果这时候引入消息队列将两个系统解耦，无论谁慢了都不会影响整体业务。

代码实现

<?php
$childList = [];
$id = ftok(__FILE__,'m');// 将路径名和项目标识符转换为 System V IPC 密钥
$msgQueue = msg_get_queue($id);//create a message queue
const MSG_TYPE = 1;
// 生产者
function producer()
{
    global $msgQueue;
    $pid = posix_getpid();// 获取当前进程 pid
    $i = rand(1,10);
    while ($i){$message = "生产者进程 ID:{$pid},round:{$i}";
        msg_send($msgQueue,MSG_TYPE,$message);
        $i--;
    }
    echo "producer success\n";
}

// 消费者
function consume()
{
    global $msgQueue;
    $pid = posix_getpid();
    while (msg_receive($msgQueue,MSG_TYPE,$msgType,1024,$message)){echo "消费者进程 ID：{$pid}，received message:{$message}\n";
    }
}

// 创建进程
function createProgress($callBack){
    global $childList;
    // 在当前进程当前位置产生分支（子进程）$pid = pcntl_fork();
    // 父进程和子进程都会执行下面代码
    if ($pid == -1){
        // 错误处理：创建子进程失败时返回 -1.
        exit('fork error');
    }elseif ($pid == 0){
        // 子进程得到的 $pid 为 0, 所以这里是子进程执行的逻辑。$pid = posix_getpid();
        echo "child progress:{$pid}\n";
        $callBack();
        exit("{$pid}:child progress exit\n");
    }else{
        // 父进程会得到子进程号，所以这里是父进程执行的逻辑
        $childList[$pid] = 1;
    }
}

createProgress('producer');
createProgress('consume');

// 等待子进程中断，防止子进程成为僵尸进程。if (!empty($childList)){$pid = pcntl_wait($status);
    if ($pid > 0){unset($childList[$pid]);
        echo "{$pid}\n";
    }
}

运行结果

环境

• pcntl 扩展：主要的进程扩展，完成进程创建于等待操作。
• sysvmsg 扩展：实现 system v 方式的进程间通信之消息队列。