swoole的process模塊創建和使用子進程

swoole中為我們提供了一個進程管理模塊 Process，替換PHP的 pcntl 擴展，方便我們創建進程，管理進程，和進程間的通信。

swoole提供了2種進程間的通信：

1、基於 unix socket 的管道 pipe。

2、基於 sysvmsg 的消息隊列。

我們可以通過 new swoole_process() 快速的創建一個進程，默認會創建一個 SOCK_DGRAM 類型的管道，用於進程間的通信，當然可以設置成其他類型，也可以不創建。

一、通過同步阻塞管道進行進程間通信

<?php

$worker_process_nums = 5;
$worker_process = [];

for ($i = 0; $i < $worker_process_nums; $i++) {
    //創建子進程
    //默認為每個子進程創建一個管道，如果不想創建設置$pipe_type參數為false
    //注意管道默認是同步阻塞，半雙工，如果讀取不到數據就會阻塞
    $worker = new swoole_process(function (swoole_process $worker) {
        //注意，如果主進程中不寫數據write()，那么子進程這里read()就會阻塞
        $task = json_decode($worker->read(), true);

        //進行計算任務
        $tmp = 0;
        for ($i = $task['start']; $i < $task['end']; $i++) {
            $tmp += $i;
        }

        echo '子進程 PID : ', $worker->pid, ' 計算 ', $task['start'], ' - ', $task['end'], ' 結果 : ', $tmp, PHP_EOL;
        //往管道中寫入計算的結果
        $worker->write($tmp);
        //子進程退出
        $worker->exit();
    });

    //保存子進程
    $worker_process[$i] = $worker;

    //啟動子進程
    $worker->start();
}

//往每個子進程管道中投遞任務
for ($i = 0; $i < $worker_process_nums; $i++) {
    $worker_process[$i]->write(json_encode([
        'start' => mt_rand(1, 10),
        'end' => mt_rand(50, 100),
    ]));
}

//父進程監聽子進程退出信號，回收子進程，防止出現僵屍進程
swoole_process::signal(SIGCHLD, function ($sig) {
    //必須為false，非阻塞模式
    while ($ret = swoole_process::wait(false)) {
        echo "子進程 PID : {$ret['pid']} 退出\n";
    }
});

　　

二、通過 swoole_event_add 將管道設為異步，來進行通信

<?php

$worker_process_nums = 5;
$worker_process = [];

for ($i = 0; $i < $worker_process_nums; $i++) {
    $worker = new swoole_process(function ($worker) {
        //在子進程中給管道添加事件監聽
        //底層會自動將該管道設置為非阻塞模式
        //參數二，是可讀事件回調函數，表示管道可以讀了
        swoole_event_add($worker->pipe, function ($pipe) use ($worker) {
            $task = json_decode($worker->read(), true);

            $tmp = 0;
            for ($i = $task['start']; $i < $task['end']; $i++) {
                $tmp += $i;
            }
            echo "子進程 : {$worker->pid} 計算 {$task['start']} - {$task['end']} \n";
            //子進程把計算的結果，寫入管道
            $worker->write($tmp);
            //注意，swoole_event_add與swoole_event_del要成對使用
            swoole_event_del($worker->pipe);
            //退出子進程
            $worker->exit();
        });
    });

    $worker_process[$i] = $worker;

    //啟動子進程
    $worker->start();
}

for ($i = 0; $i < $worker_process_nums; $i++) {
    $worker = $worker_process[$i];

    $worker->write(json_encode([
        'start' => mt_rand(1, 10),
        'end' => mt_rand(50, 100),
    ]));

    //主進程中，監聽子進程管道事件
    swoole_event_add($worker->pipe, function ($pipe) use ($worker) {
        $result = $worker->read();
        echo "子進程 : {$worker->pid} 計算結果 {$result} \n";
        swoole_event_del($worker->pipe);
    });
}

//父進程監聽子進程退出信號，回收子進程，防止出現僵屍進程
swoole_process::signal(SIGCHLD, function ($sig) {
    //必須為false，非阻塞模式
    while ($ret = swoole_process::wait(false)) {
        echo "子進程 PID : {$ret['pid']} 退出\n";
    }
});

　　

三、使用消息隊列來完成進程間通信

<?php

$worker_process_nums = 5;
$worker_process = [];

for ($i = 0; $i < $worker_process_nums; $i++) {
    //注意，這里將參數$pipe_type設為false，表示不創建管道
    $worker = new swoole_process(function ($worker) {
        $task = json_decode($worker->pop(), true);

        $tmp = 0;
        for ($i = $task['start']; $i < $task['end']; $i++) {
            $tmp += $i;
        }
        echo "子進程 : {$worker->pid} 計算 {$task['start']} - {$task['end']} \n";
        $worker->push($tmp);
        $worker->exit();
    }, false, false);

    //使用消息隊列，作為進程間的通信
    //注意，消息隊列是共享的
    $worker->useQueue();

    $worker_process[$i] = $worker;

    //啟動子進程
    $worker->start();
}

for ($i = 0; $i < $worker_process_nums; $i++) {
    //只需用一個子進程發送消息即可，因為消息隊列是共享的
    $worker_process[0]->push(json_encode([
        'start' => mt_rand(1, 10),
        'end' => mt_rand(50, 100),
    ]));
}

//注意，這里要暫停，防止加入隊列的任務，立刻被主進程讀出來。
sleep(1);

for ($i = 0; $i < $worker_process_nums; $i++) {
    $result = $worker_process[0]->pop();
    echo "計算結果 : {$result} \n";
}

//父進程監聽子進程退出信號，回收子進程，防止出現僵屍進程
swoole_process::signal(SIGCHLD, function ($sig) {
    //必須為false，非阻塞模式
    while ($ret = swoole_process::wait(false)) {
        echo "子進程 PID : {$ret['pid']} 退出\n";
    }
});

　　

四、進程可以通過 signal 監聽信號，和 alarm 設置定時器。

我們可以在父進程上設置監聽信號，當子進程退出時，重新掛起子進程。

也可以設置定時器，通過 swoole_process::kill($pid, 0); 定時檢測進程是否存活。

<?php

//每隔1秒觸發SIGALAM信號
//注意，alarm不能和Timer同時使用
swoole_process::alarm(1000 * 1000, 0);

swoole_process::signal(SIGALRM, function ($signo) {
    static $cnt = 0;
    $cnt++;
    echo "時鍾定時信號\n";

    if ($cnt > 10) {
        //清除定時器
        swoole_process::alarm(-1);
    }
});

swoole_process::signal(SIGINT, function ($signo) {
    echo "我被ctrl+c了\n";

    //退出主進程，不然將一直無法正常退出
    exit(0);
});