PHP のマルチプロセス消費キューについての話

はじめに

最近、キュー mcq を使用してプロセス消費キュー データを開始する小さな関数を開発しました。その後、処理できないことが判明し、別の処理を追加しましたが、しばらくすると再び処理できなくなりました...

この方法では、毎回 crontab を変更する必要があります。ハングし、開始が間に合わず、次回 crontab が実行されるまで開始されません。 Kill はプロセスを閉じる (再起動する) ときに使用され、処理中のデータが失われる可能性があります。たとえば、次の例では、処理ロジックがスリープ プロセスであると仮定します。効果を明確に確認するために、処理時間は10 秒に拡大:

<?php
$i = 1;
while (1) {
    echo "开始第[{$i}]次循环\n";
    sleep(10);
    echo "结束第[{$i}]次循环\n";
    $i++;
}

スクリプトを実行した後、ループが開始されるまで待ってから、kill {$pid} をプロセスに送信します。デフォルトでは、SIGTERM 15 番の信号が送信されます。 $i がキューから取得されたとします。それが 2 になると、処理中です。プログラムに kill シグナルを送信します。キューのデータ損失と同様に、問題は比較的大きいため、次のようにする必要があります。これらの問題を解決する方法を見つけてください。

开始第[1]次循环
结束第[1]次循环
开始第[2]次循环


[1]    28372 terminated  php t.php

nginx プロセス モデル

このとき、nginx を思い出しました。nginx は高性能サーバーの主力として、数千ものサーバーにサービスを提供しています個人サービスの場合、彼のプロセス モデルは、以下に示すように、より古典的です:

管理者は、/path/ からマスター プロセスを通じて nginx と対話します。 to/nginx .pidnginx マスター プロセスの pid を読み取り、マスター プロセスにシグナルを送信します。マスターはさまざまなシグナルに応じてさまざまな処理を実行し、管理者に情報をフィードバックします。ワーカーはマスター プロセスからフォークされます。マスターはワーカーの管理を担当し、ビジネスは処理しません。ワーカーは特定のビジネスのハンドラーです。マスターはワーカーの終了と起動を制御できます。ワーカーが予期せず終了した場合子プロセスが終了したというメッセージをマスターが受信すると、新しいワーカー プロセスが再起動されて補完され、業務処理には影響しません。 nginx は、処理中のデータを失うことなくスムーズに終了することもでき、構成を更新するときに、nginx はオンライン サービスに影響を与えることなく新しい構成をロードできるため、リクエスト量が多い場合に特に役立ちます。

プロセス設計

nginx の高度なモデルを確認した後、mcq データの処理のニーズを満たす同様のクラス ライブラリを開発できます。単一のファイルですべてのプロセスを制御し、スムーズに終了し、子プロセスのステータスを表示できます。キューデータの受信にある程度の遅延が発生する場合に対処するため、それほど複雑にする必要はありませんが、nginx のような中断のないサービスを提供するのは面倒で時間と労力がかかり、あまり意味がありません。設計されたプロセス モデルは nginx に似ており、nginx の簡易バージョンに似ています。

プロセス セマフォの設計

セマフォはプロセス間の通信方法です。比較的単純で、弱い単一の機能しかありません。プロセスにシグナルを送信すると、プロセスはシグナルに応じてさまざまな処理を実行します。

マスター プロセスが開始されたら、pid をファイル /path/to/daeminze.pid に保存します。管理者はシグナルを通じてマスター プロセスと通信します。マスター プロセスは 3 種類の以下に示すように、シグナルは異なる方法で処理されます:

SIGINT 	=> 平滑退出，处理完正在处理的数据再退出
SIGTERM => 暴力退出，无论进程是否正在处理数据直接退出
SIGUSR1 => 查看进程状态，查看进程占用内存，运行时间等信息

マスター プロセスは、シグナルを通じてワーカー プロセスと通信します。ワーカー プロセスは、以下に示すように 2 つのシグナルをインストールしました:

SIGINT 	=> 平滑退出
SIGUSR1	=> 查看worker进程自身状态

ワーカー プロセスが存在する理由 シグナルは 2 つだけインストールされており、そのうちの 1 つは SIGTERM が欠落しています。これは、マスター プロセスがシグナル SIGTERM を受信した後、SIGKILL シグナルをワーカー プロセス、およびプロセスはデフォルトで強制的に閉じられます。

ワーカー プロセスはマスター プロセスによってフォークアウトされるため、マスター プロセスは pcntl_wait を通じて子プロセスの終了イベントを待つことができます。子プロセスが終了すると、子プロセスの pid は次のようになります。返され、処理され、新しいプロセスを開始して追加します。

マスター プロセスも、pcntl_wait を介してシグナルの受信を待機します。シグナルが到着すると、-1 が返されます。この場所にはまだいくつかの落とし穴があります。詳細は後述します。

PHP でシグナルをトリガーするには 2 つの方法があります。1 つ目の方法は declare(ticks = 1); です。これは効率的ではありません。Zend が低レベルのステートメントを実行するたびに、 will go プロセス内に未処理のシグナルがあるかどうかを確認します。現在はほとんど使用されていません。PHP 5.3.0 およびそれ以前のバージョンでは、これが使用される可能性があります。

2 番目の方法は、pcntl_signal_dispatch を通じて未処理のシグナルを呼び出すことです。PHP 5.4.0 以降のバージョンが適用可能です。この関数はループ内にうまく配置できます。基本的にパフォーマンスの低下はなく、現時点での使用をお勧めします。

PHP はシグナルをインストールおよび修復します signal

PHP は pcntl_signal を通じてシグナルをインストールします。関数の宣言は次のとおりです:

bool pcntl_signal ( int $signo , [callback $handler [, bool $restart_syscalls = true ] )

第三个参数restart_syscalls不太好理解，找了很多资料，也没太查明白，经过试验发现，这个参数对pcntl_wait函数接收信号有影响，当设置为缺省值true的时候，发送信号，进程用pcntl_wait收不到，必须设置为false才可以，看看下面这个例子：

<?php
$i = 0;
while ($i<5) {
    $pid = pcntl_fork();
    $random = rand(10, 50);
    if ($pid == 0) {
        sleep($random);
        exit();
    }
    echo "child {$pid} sleep {$random}\n";
    $i++;
}

pcntl_signal(SIGINT,  function($signo) {
     echo "Ctrl + C\n";
});

while (1) {
    $pid = pcntl_wait($status);
    var_dump($pid);
    pcntl_signal_dispatch();
}

运行之后，我们对父进程发送kill -SIGINT {$pid}信号，发现pcntl_wait没有反应，等到有子进程退出的时候，发送过的SIGINT会一个个执行，比如下面结果：

child 29643 sleep 48
child 29644 sleep 24
child 29645 sleep 37
child 29646 sleep 20
child 29647 sleep 31
int(29643)
Ctrl + C
Ctrl + C
Ctrl + C
Ctrl + C
int(29646)

这是运行脚本之后马上给父进程发送了四次SIGINT信号，等到一个子进程推出的时候，所有信号都会触发。

但当把安装信号的第三个参数设置为false：

pcntl_signal(SIGINT,  function($signo) {
     echo "Ctrl + C\n";
}, false);

这时候给父进程发送SIGINT信号，pcntl_wait会马上返回-1，信号对应的事件也会触发。

所以第三个参数大概意思就是，是否重新注册此信号，如果为false只注册一次，触发之后就返回，pcntl_wait就能收到消息，如果为true，会重复注册，不会返回，pcntl_wait收不到消息。

信号量和系统调用

信号量会打断系统调用，让系统调用立刻返回，比如sleep，当进程正在sleep的时候，收到信号，sleep会马上返回剩余sleep秒数，比如：

运行之后，按Ctrl + C，结果如下所示：
123
^Climit sleep [1] s
Ctrl + C
123
limit sleep [0] s
123
^Climit sleep [1] s
Ctrl + C
123
^Climit sleep [2] s
daemon（守护）进程
这种进程一般设计为daemon进程，不受终端控制，不与终端交互，长时间运行在后台，而对于一个进程，我们可以通过下面几个步骤把他升级为一个标准的daemon进程：
protected function daemonize()
{
    $pid = pcntl_fork();
    if (-1 == $pid) {
        throw new Exception("fork进程失败");
    } elseif ($pid != 0) {
        exit(0);
    }
    if (-1 == posix_setsid()) {
        throw new Exception("新建立session会话失败");
    }

    $pid = pcntl_fork();
    if (-1 == $pid) {
        throw new Exception("fork进程失败");
    } else if($pid != 0) {
        exit(0);
    }

    umask(0);
    chdir("/");
}
拢共分五步：
1、fork子进程，父进程退出。
2、设置子进程为会话组长，进程组长。
3、再次fork，父进程退出，子进程继续运行。
4、恢复文件掩码为0。
5、切换当前目录到根目录/。
第2步是为第1步做准备，设置进程为会话组长，必要条件是进程非进程组长，因此做第一次fork，进程组长（父进程）退出，子进程通过posix_setsid()设置为会话组长，同时也为进程组长。
第3步是为了不让进程重新控制终端，因为一个进程控制一个终端的必要条件是会话组长（pid=sid）。
第4步是为了恢复默认的文件掩码，避免之前做的操作对文件掩码做了设置，带来不必要的麻烦。关于文件掩码， linux中，文件掩码在创建文件、文件夹的时候会用到，文件的默认权限为666，文件夹为777，创建文件（夹）的时候会用默认值减去掩码的值作为创建文件（夹）的最终值，比如掩码022下创建文件666 - 222 = 644，创建文件夹777 - 022 = 755：


掩码
新建文件权限
新建文件夹权限



umask(0)
666 (-rw-rw-rw-)
777 (drwxrwxrwx)


umask(022)
644 (-rw-r--r--)
755 (drwxr-xr-x)



第5步是切换了当前目录到根目录/，网上说避免起始运行他的目录不能被正确卸载，这个不是太了解。
对应5步，每一步的各种id变化信息：


操作后
pid
ppid
pgid
sid



开始
17723
31381
17723
31381


第一次fork
17723
1
17723
31381


posix_setsid()
17740
1
17740
17740


第二次fork
17840
1
17740
17740



另外，会话、进程组、进程的关系如下图所示，这张图有助于更好的理解。

至此，你也可以轻松地造出一个daemon进程了。
命令设计
我准备给这个类库设计6个命令，如下所示：

start 启动命令
restart 强制重启
stop 平滑停止
reload 平滑重启
quit 强制停止
status 查看进程状态

启动命令
启动命令就是默认的流程，按照默认流程走就是启动命令，启动命令会检测pid文件中是否已经有pid，pid对应的进程是否健康，是否需要重新启动。
强制停止命令
管理员通过入口文件结合pid给master进程发送SIGTERM信号，master进程给所有子进程发送SIGKILL信号，等待所有worker进程退出后，master进程也退出。
强制重启命令
强制停止命令 + 启动命令
平滑停止命令
平滑停止命令，管理员给master进程发送SIGINT信号，master进程给所有子进程发送SIGINT，worker进程将自身状态标记为stoping，当worker进程下次循环的时候会根据stoping决定停止，不在接收新的数据，等所有worker进程退出之后，master进程也退出。
平滑重启命令
平滑停止命令 + 启动命令
查看进程状态
查看进程状态这个借鉴了workerman的思路，管理员给master进程发送SIGUSR1信号，告诉主进程，我要看所有进程的信息，master进程，master进程将自身的进程信息写入配置好的文件路径A中，然后发送SIGUSR1，告诉worker进程把自己的信息也写入文件A中，由于这个过程是异步的，不知道worker进程啥时候写完，所以master进程在此处等待，等所有worker进程都写入文件之后，格式化所有的信息输出，最后输出的内容如下所示：
/dir /usr/local/bin/php DaemonMcn.php status
Daemon [DaemonMcn] 信息:
-------------------------------- master进程状态 --------------------------------
pid       占用内存       处理次数       开始时间                 运行时间
16343     0.75M          --             2018-05-15 09:42:45      0 天 0 时 3 分
12 slaver
-------------------------------- slaver进程状态 --------------------------------
任务task-mcq:
16345     0.75M          236            2018-05-15 09:42:45      0 天 0 时 3 分
16346     0.75M          236            2018-05-15 09:42:45      0 天 0 时 3 分
--------------------------------------------------------------------------------
任务test-mcq:
16348     0.75M          49             2018-05-15 09:42:45      0 天 0 时 3 分
16350     0.75M          49             2018-05-15 09:42:45      0 天 0 时 3 分
16358     0.75M          49             2018-05-15 09:42:45      0 天 0 时 3 分
16449     0.75M          1              2018-05-15 09:46:40      0 天 0 时 0 分
--------------------------------------------------------------------------------
等待worker进程将进程信息写入文件的时候，这个地方用了个比较trick的方法，每个worker进程输出一行信息，统计文件的行数，达到worker进程的行数之后表示所有worker进程都将信息写入完毕，否则，每个1s检测一次。
其他设计
另外还加了两个比较实用的功能，一个是worker进程运行时间限制，一个是worker进程循环处理次数限制，防止长时间循环进程出现内存溢出等意外情况。时间默认是1小时，运行次数默认是10w次。
除此之外，也可以支持多任务，每个任务几个进程独立开，统一由master进程管理。
代码已经放到github中，有兴趣的可以试试，不支持windows哦，有什么错误还望指出来。
相关教程推荐：《PHP教程》
以上がPHP のマルチプロセス消費キューについての話の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。