Sprechen Sie über die Multiprozess-Verbrauchswarteschlange in PHP

Einführung

Kürzlich eine kleine Funktion entwickelt, die queue mcq verwendet, um einen Prozess zu starten Verbrauchswarteschlange Daten, später wurde festgestellt, dass ein Prozess nicht verarbeitet werden konnte und ein anderer Prozess hinzugefügt wurde. Nach einer Weile konnte er nicht erneut verarbeitet werden ...

Diese Methode erfordert jedes Mal eine Änderung der Crontab. Wenn der Prozess hängt, wird er erst bei der nächsten Ausführung von crontab gestartet. Kill wird beim Schließen (Neustarten) des Prozesses verwendet, wodurch die verarbeiteten Daten verloren gehen können. Im folgenden Beispiel gehen wir davon aus, dass der Ruheprozess die Verarbeitungslogik ist Auf 10 Sekunden vergrößert:

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

Nachdem wir das Skript ausgeführt haben, warten Sie, bis die Schleife beginnt, und senden Sie kill {$pid} an den Prozess. Standardmäßig wird das SIGTERM-Signal mit der Nummer 15 gesendet. Angenommen, $i wird aus der Warteschlange abgerufen. Wenn es 2 wird, senden wir ein Kill-Signal an das Programm. Das Problem ist relativ groß, also muss ich einen Weg finden diese Probleme lösen.

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


[1]    28372 terminated  php t.php

Nginx-Prozessmodell

Zu diesem Zeitpunkt dachte ich an Nginx als Hauptstütze von Hochleistungsservern, das Tausende von Unternehmen bedient Für persönliche Dienste ist das Prozessmodell klassischer, wie unten gezeigt:

Der Administrator interagiert mit Nginx über den Masterprozess und liest den Nginx-Masterprozess aus /path/to/nginx.pid pid, sendet ein Signal an den Master-Prozess, der Master führt je nach Signal unterschiedliche Verarbeitungen durch und gibt die Informationen dann an den Administrator zurück. Worker werden vom Master-Prozess ausgeschlossen. Der Master ist für die Verwaltung der Worker verantwortlich und kümmert sich nicht um die Abwicklung bestimmter Geschäfte Der Master erhält eine Benachrichtigung, dass der untergeordnete Prozess beendet wurde. Neue Arbeitsprozesse werden neu gestartet, um sie zu ergänzen, sodass die Geschäftsverarbeitung nicht beeinträchtigt wird. Nginx kann auch reibungslos beendet werden, ohne dass verarbeitete Daten verloren gehen. Beim Aktualisieren der Konfiguration kann Nginx die neue Konfiguration laden, ohne den Onlinedienst zu beeinträchtigen, was besonders nützlich ist, wenn das Anforderungsvolumen groß ist.

Prozessdesign

Nachdem wir uns das erweiterte Modell von Nginx angesehen haben, können wir eine ähnliche Klassenbibliothek entwickeln, um den Anforderungen der Verarbeitung von MCQ-Daten gerecht zu werden Eine einzelne Datei kann alle Prozesse steuern, reibungslos beenden und den Status untergeordneter Prozesse anzeigen. Es muss nicht allzu kompliziert sein, da wir mit einer gewissen Verzögerung beim Empfang von Warteschlangendaten zu kämpfen haben. Es ist mühsam, zeitaufwändig und arbeitsintensiv, einen unterbrechungsfreien Dienst wie Nginx bereitzustellen, und es ist nicht sehr sinnvoll. Das entworfene Prozessmodell ähnelt Nginx, eher einer vereinfachten Version von Nginx.

Prozess-Semaphor-Design

Semaphor ist eine Art der Kommunikation zwischen Prozessen. Es ist relativ einfach und hat nur eine schwache Einzelfunktion Senden Sie ein Signal an den Prozess, und der Prozess führt je nach Signal unterschiedliche Verarbeitungen durch.

Wenn der Master-Prozess startet, speichern Sie die PID in der Datei /path/to/daeminze.pid Der Administrator kommuniziert mit dem Master-Prozess über Signale. Der Master-Prozess installiert 3 Arten von Signalen Verarbeitung, wie unten gezeigt:

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

Der Master-Prozess kommuniziert mit dem Worker-Prozess über Signale. Der Worker-Prozess hat 2 Signale installiert, wie unten gezeigt:

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

Warum wird der Worker-Prozess nur installiert? 2 Signale? Eines fehltSIGTERM, denn nachdem der Master-Prozess das Signal SIGTERM empfängt, sendet er das Signal SIGKILL an den Worker-Prozess und der Prozess kann standardmäßig zum Schließen gezwungen werden.

Der Arbeitsprozess wird über den Master-Prozess weitergeleitet, sodass der Master-Prozess bis pcntl_wait auf das Exit-Ereignis des untergeordneten Prozesses warten kann. Wenn ein untergeordneter Prozess beendet wird, gibt er die PID des untergeordneten Prozesses zurück und führt die Verarbeitung durch und startet einen neuen Prozess.

Der Master-Prozess wartet auch auf den Empfang von Signalen über pcntl_wait. Wenn ein Signal eintrifft, wird es zurückgegeben -1. An dieser Stelle gibt es noch einige Fallstricke, die weiter unten ausführlich besprochen werden.

Es gibt zwei Möglichkeiten, Signale in PHP auszulösen. Die erste Methode ist declare(ticks = 1);, die nicht effizient ist. Jedes Mal, wenn Zend eine Low-Level-Anweisung ausführt, prüft es, ob im Prozess nicht verarbeitete Signale vorhanden sind . Es wird derzeit nur noch selten verwendet PHP 5.3.0 und frühere Versionen verwenden dies möglicherweise.

Die zweite Methode besteht darin, unverarbeitete Signale über pcntl_signal_dispatch aufzurufen, was auf PHP 5.4.0 und spätere Versionen anwendbar ist. Sie können diese Funktion geschickt in einer Schleife platzieren, und es gibt grundsätzlich keinen Leistungsverlust wird ab sofort zum Einsatz empfohlen.

PHP installiert und repariert das Semaphor

PHP installiert das Signal über pcntl_signal Die Funktionsdeklaration lautet wie folgt:

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秒数，比如：

Nach dem Login kopieren

运行之后，按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个命令，如下所示：

1. start 启动命令

2. restart 强制重启

3. stop 平滑停止

4. reload 平滑重启

5. quit 强制停止

6. 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教程》

Das obige ist der detaillierte Inhalt vonSprechen Sie über die Multiprozess-Verbrauchswarteschlange in PHP. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!