谈谈PHP中的多进程消费队列-百分百源码网

谈谈PHP中的多进程消费队列

发布时间：09/01 来源：未知浏览：关键词：

引言

比来开发一个小功效，用到了队列mcq，启动一个进程消耗队列数据，后边发明一个进程处置不外来了，又加了一个进程，过了段时间又处置不外来了......

这种方式每次都要修改crontab，假如进程挂掉了，不会及时的启动，要比及下次crontab施行的时候才会启动。关闭（重新启动）进程的时候用的是kill，这大概会丧失正在处置的数据，比方下面这个例子，我们假设sleep历程就是处置逻辑，这里为了明显看出结果，将处置时间放大到10s：

<?php
$i = 1;
while (1) {
    echo "开端第[{$i}]次轮回\n";
    sleep(10);
    echo "完毕第[{$i}]次轮回\n";
    $i++;
}

当我们运转足本之后，比及轮回开端之后，给进程发送 kill {$pid}，默许发送的是编号为15的SIGTERM信号。假设$i是从队列拿到的，拿到2的时候，正在处置，我们给程序发送了kill信号，和队列数据丧失一样，问题比力大，因此我要想方法解决这些问题。

开端第[1]次轮回
完毕第[1]次轮回
开端第[2]次轮回


[1]    28372 terminated  php t.php

nginx进程模型

这时候我想到了nginx，nginx作为高机能效劳器的国家栋梁，为成千上万的公司和个人效劳，他的进程模型比力经典，如下所示：

nginx 进程模型图片来自网络

治理员通过master进程和nginx停止交互，从/path/to/nginx.pid读取nginx master进程的pid，发送信号给master进程，master按照不一样的信号做出不一样的处置，然后反应信息给治理员。worker是master进程fork出来的，master负责治理worker，不会去处置业务，worker才是详细业务的处置者，master可以操纵worker的退出、启动，当worker不测退出，master会收到子进程退出的新闻，也会从新启动新的worker进程补充上来，不让业务处置受影响。nginx还可以平滑退出，不丧失任何一个正在处置的数据，更新配置时nginx可以做到不影响线上效劳来加载新的配置，这在恳求量很大的时候特殊有用。

进程设计

看了nginx的进模型，我们完全可以开发一个相似的类库来知足处置mcq数据的需求，做到单文件操纵所有进程、可以平滑退出、可以查看子进程状态。不需要太复杂，由于我们处置队列数据接收必然的延迟，做到nginx那样不中断效劳比力费事，费时费劲，意义不是很大。设计的进程模型跟nginx相似，更像是nginx的简化版本。
多进程模型

进程信号量设计

信号量是进程间通讯的一种方式，比力简便，单功效也比力弱，只能发送信号给进程，进程按照信号做出不一样的处置。

master进程启动的时候留存pid到文件/path/to/daeminze.pid，治理员通过信号和master进程通讯，master进程安置3种信号，碰到不一样的信号，做出不一样的处置，如下所示：

SIGINT 	=> 平滑退出，处置完正在处置的数据再退出
SIGTERM => 暴力退出，不管进程可否正在处置数据直接退出
SIGUSR1 => 查看进程状态，查看进程占用内存，运转时间等信息

master进程通过信号和worker进程通讯，worker进程安置了2个信号，如下所示：

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

为什么worker进程只安置2个信号呢，少了个SIGTERM，由于master进程收到信号SIGTERM之后，向worker进程发送SIGKILL信号，默许强迫关闭进程即可。

worker进程是通过master进程fork出来的，这样master进程可以通过pcntl_wait来等候子进程退出事件，当有子进程退出的时候返回子进程pid，做处置并启动新的进程补充上来。

master进程也通过pcntl_wait来等候接收信号，当有信号抵达的时候，会返回-1，这个地方还有些坑，鄙人文中会具体讲。

PHP中有2种信号触发的方式，第一种方式是declare(ticks = 1);，这种效力不高，Zend每施行一次初级语句，都会去检查进程中可否有未处置的信号，此刻已经很少使用了，PHP 5.3.0及此前的版本大概会用到这个。

第二种是通过pcntl_signal_dispatch来调取未处置的信号，PHP 5.4.0及之后的版本适用，可以奇妙的将该函数放在轮回中，机能上根本没什么亏损，此刻引荐适用。

PHP安置修信号量

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

<?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个命令，如下所示：

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