Une brève discussion sur les files d'attente de consommation multi-processus en PHP

Cet article vous présente la file d'attente de consommation multi-processus en PHP. Il a une certaine valeur de référence. Les amis dans le besoin peuvent s'y référer. J'espère qu'il sera utile à tout le monde.

Étude recommandée : "Tutoriel vidéo PHP"

Récemment, une petite fonction a été développée, en utilisant queue mcq, pour démarrer un processus pour la consommation des données de la file d'attente, plus tard, il a été constaté qu'un processus ne pouvait pas être traité, et un autre processus a été ajouté après un certain temps, il n'a pas pu être traité à nouveau...

Cette méthode nécessite de modifier le. crontab à chaque fois. Si le processus se bloque et ne démarre pas à temps, il ne démarrera pas avant la prochaine exécution de crontab. Kill est utilisé lors de la fermeture (du redémarrage) du processus, ce qui peut entraîner la perte des données en cours de traitement. Par exemple, dans l'exemple suivant, nous supposons que le processus de veille est la logique de traitement. Afin de voir clairement l'effet, le temps de traitement est. agrandi à 10s :

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

Après avoir exécuté le script, attendez que la boucle démarre et envoyez kill {$pid} au processus. Par défaut, le signal SIGTERM numéroté 15 est envoyé. Supposons que $i soit obtenu à partir de la file d'attente. Lorsqu'il obtient 2, il est en cours de traitement. Comme pour la perte de données dans la file d'attente, le problème est relativement important, je dois donc trouver un moyen de le faire. résoudre ces problèmes.

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


[1]    28372 terminated  php t.php

modèle de processus nginx

À cette époque, je pensais à nginx En tant que pilier des serveurs hautes performances, nginx sert des milliers d'entreprises et de particuliers. Son modèle de processus est relativement classique, comme suit. Montré :

L'administrateur interagit avec nginx via le processus maître, lit le pid du processus maître nginx à partir de /path/to/nginx.pid et envoie un signal au processus maître. Le maître traite le signal. Le signal est traité différemment, puis les informations sont renvoyées à l'administrateur. Les travailleurs sont issus du processus maître. Le maître est responsable de la gestion des travailleurs et ne gérera pas les affaires. Le travailleur est le gestionnaire d'une activité spécifique. Le maître peut contrôler la sortie et le démarrage du travailleur. master recevra une notification indiquant que le processus enfant est terminé, les nouveaux processus de travail seront redémarrés pour les compléter, afin que le traitement métier ne soit pas affecté. nginx peut également se terminer en douceur sans perdre aucune donnée en cours de traitement. Lors de la mise à jour de la configuration, nginx peut charger la nouvelle configuration sans affecter le service en ligne, ce qui est particulièrement utile lorsque le volume de requêtes est important.

Conception de processus

Après avoir examiné le modèle avancé de nginx, nous pouvons développer une bibliothèque de classes similaire pour répondre aux besoins de traitement des données mcq, afin qu'un seul fichier puisse contrôler tous les processus et permettre une sortie en douceur . Vous pouvez afficher l'état des processus enfants. Cela n'a pas besoin d'être trop compliqué, car nous sommes confrontés à un certain retard dans la réception des données de la file d'attente. Il est fastidieux, long et laborieux de fournir un service ininterrompu comme nginx, et ce n'est pas très significatif. Le modèle de processus conçu est similaire à nginx, ressemblant davantage à une version simplifiée de nginx.

Conception de sémaphore de processus

Le sémaphore est un moyen de communication entre les processus. Il est relativement simple et a une fonction unique faible. Il ne peut envoyer des signaux qu'aux processus. Différents traitements sont effectués en fonction du signal.

Lorsque le processus maître démarre, enregistrez le pid dans le fichier /path/to/daeminze.pid. L'administrateur communique avec le processus maître via des signaux. Le processus maître installe 3 types de signaux. Lorsqu'il rencontre différents signaux, différents processus sont effectués. , comme indiqué ci-dessous :

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

Le processus maître communique avec le processus de travail via des signaux. Le processus de travail a installé 2 signaux, comme indiqué ci-dessous :

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

Pourquoi le processus de travail n'en installe-t-il que 2. signaux ? Il en manque unSIGTERM, car une fois que le processus maître a reçu le signal SIGTERM, il envoie le signal SIGKILL au processus de travail, et le processus peut être forcé à se fermer par défaut.

Le processus de travail est réparti via le processus maître, de sorte que le processus maître peut attendre l'événement de sortie du processus enfant via pcntl_wait Lorsqu'un processus enfant se termine, il renvoie le pid du processus enfant, effectue le traitement et. démarre un nouveau processus.

Le processus maître attend également de recevoir des signaux via pcntl_wait Lorsqu'un signal arrive, il reviendra -1 Il y a encore quelques pièges à cet endroit, qui seront discutés en détail ci-dessous.

Il existe deux façons de déclencher des signaux en PHP. La première est declare(ticks = 1);, ce qui n'est pas efficace. Chaque fois que Zend exécute une instruction de bas niveau, il vérifiera s'il y a des signaux non gérés dans le processus. . Il est rarement utilisé maintenant PHP 5.3.0 et les versions précédentes peuvent l'utiliser.

La deuxième méthode consiste à appeler des signaux non traités via pcntl_signal_dispatch, ce qui est applicable aux versions PHP 5.4.0 et ultérieures. Vous pouvez intelligemment placer cette fonction dans une boucle, et il n'y a fondamentalement aucune perte de performances. est recommandé pour une utilisation maintenant.

PHP installe et répare le sémaphore

PHP installe le signal via pcntl_signal, la déclaration de la fonction est la suivante :

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

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

while (true) {
	pcntl_signal_dispatch();
    echo "123\n";
    $limit = sleep(2);
	echo "limit sleep [{$limit}] s\n";
}

运行之后，按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哦，有什么错误还望指出来。

更多编程相关知识，请访问：编程入门！！

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!