背景
由于PHP不支持多线程&＃xff0c;但是作为一个完善的系统&＃xff0c;有很多操作都是需要异步完成的。为了完成这些异步操作&＃xff0c;我们做了一个基于Redis队列任务系统。
大家知道&＃xff0c;一个消息队列处理系统主要分为两大部分&＃xff1a;消费者和生产者。
在我们的系统中&＃xff0c;主系统作为生产者&＃xff0c;任务系统作为消费者。
具体的工作流程如下&＃xff1a;
1、主系统将需要需要处理的任务名称&＃43;任务参数push到队列中。
2、任务系统实时的对任务队列进行pop&＃xff0c;pop出来一个任务就fork一个子进程&＃xff0c;由子进程完成具体的任务逻辑。
具体代码如下&＃xff1a;
/*** 启动守护进程*/
public function runAction() {Tools::log_message(&＃39;ERROR&＃39;, &＃39;daemon/run&＃39; . &＃39; | action: restart&＃39;, &＃39;daemon-&＃39;);while (true) {$this->fork_process();}exit;
}/*** 创建子进程*/
private function fork_process() {$ppid &＃61; getmypid();$pid &＃61; pcntl_fork();if ($pid &＃61;&＃61; 0) {//子进程$pid &＃61; posix_getpid();//echo "* Process {$pid} was created nn";$this->mq_process();exit;} else {//主进程$pid &＃61; pcntl_wait($status, WUNTRACED); //取得子进程结束状态if (pcntl_wifexited($status)) {//echo "nn* Sub process: {$pid} exited with {$status}";//Tools::log_message(&＃39;INFO&＃39;, &＃39;daemon/run succ&＃39; . &＃39;|status:&＃39; . $status . &＃39;|pid:&＃39; . $ppid . &＃39;|childpid:&＃39; . $pid );} else {Tools::log_message(&＃39;ERROR&＃39;, &＃39;daemon/run fail&＃39; . &＃39;|status:&＃39; . $status . &＃39;|pid:&＃39; . $ppid . &＃39;|childpid:&＃39; . $pid, &＃39;daemon-&＃39;);}}
}/*** 业务任务队列处理*/
private function mq_process() {$data_pop &＃61; $this->masterRedis->rPop($this->redis_list_key);$data &＃61; json_decode($data_pop, 1);if (!$data) {return FALSE;}$worker &＃61; &＃39;_task_&＃39; . $data[&＃39;worker&＃39;];$class_name &＃61; isset($data[&＃39;class&＃39;]) ? $data[&＃39;class&＃39;] : &＃39;TaskproModel&＃39;;$params &＃61; $data[&＃39;params&＃39;];$class &＃61; new $class_name();$class->$worker($params);return TRUE;
}
这是一个简单的任务处理系统。通过这个任务系统帮助我们实现了异步&＃xff0c;到目前为止已经稳定运行了将近一年。但很可惜&＃xff0c;它是一个单进程的系统。它是一直在不断的fork&＃xff0c;如果有任务就处理&＃xff0c;没有任务就跳过&＃xff0c;这样很稳定。
但问题有两个&＃xff1a;一是不断地fork、pop会浪费服务器资源&＃xff0c;二是不支持并发&＃xff01;第一个问题还好&＃xff0c;但第二个问题就很严重。当主系统 同时 抛过来大量的任务时&＃xff0c;任务的处理时间就会无限的拉长。
新的设计
为了解决并发的问题&＃xff0c;我们计划做一个更加高效强壮的队里处理系统。
因为在PHP7之前不支持多线程&＃xff0c;所以我们采用多进程。
找了不少资料&＃xff0c;大多所谓的多进程都是N个进程同时在后台运行。
显然这是不合适的。
我的预想是&＃xff1a;每pop出一个任务就fork一个任务&＃xff0c;任务执行完成后子进程结束。
遇到的问题
1、如何控制最大进程数
这个问题很简单&＃xff0c;那就是每fork一个子进程就自增一次。而当子进程执行完成就自减一次。
自增没有问题&＃xff0c;我们就在主进程中操作就完了。那么该如何自减呢&＃xff1f;
可能你会说&＃xff0c;当然是在子进程中啊。但这里你需要注意&＃xff1a;当fork的时候是从主进程复制了一份资源给子进程&＃xff0c;这就意味着你无法在子进程中操作主进程中的计数器&＃xff01;
所以&＃xff0c;这里就需要了解一个知识点&＃xff1a;信号。
具体的可以自行Google&＃xff0c;这里直接看代码。
// install signal handler for dead kids
pcntl_signal(SIGCHLD, array($this, "sig_handler"));
这就安装了一个信号处理器。当然还缺少一点。
declare是一个控制结构语句&＃xff0c;具体的用法也请去Google。
这句代码的意思就是每执行一条低级语句就调用一次信号处理器。
这样&＃xff0c;每当子进程结束的时候就会调用信号处理器&＃xff0c;我们就可以在信号处理器中进行自减。
2、如何解决进程残留
在多进程开发中&＃xff0c;如果处理不当就会导致进程残留。
为了解决进程残留&＃xff0c;必须得将子进程回收。
那么如何对子进程进行回收就是一个技术点了。
在pcntl的demo中&＃xff0c;包括很多博文中都是说在主进程中回收子进程。
但我们是基于Redis的brpop的&＃xff0c;而brpop是阻塞的。
这就导致一个问题&＃xff1a;当执行N个任务之后&＃xff0c;任务系统空闲的时候主进程是阻塞的&＃xff0c;而在发生阻塞的时候子进程还在执行&＃xff0c;所以就无法完成最后几个子进程的进程回收。。。
这里本来一直很纠结&＃xff0c;但当我将信号处理器搞定之后就也很简单了。
进程回收也放到信号处理器中去。
新系统的评估
pcntl是一个进程处理的扩展&＃xff0c;但很可惜它对多进程的支持非常乏力。
所以这里采用Swoole扩展中的Process。
具体代码如下&＃xff1a;
declare(ticks &＃61; 1);
class JobDaemonController extends Yaf_Controller_Abstract{use Trait_Redis;private $maxProcesses &＃61; 800;private $child;private $masterRedis;private $redis_task_wing &＃61; &＃39;task:wing&＃39;; //待处理队列public function init(){// install signal handler for dead kidspcntl_signal(SIGCHLD, array($this, "sig_handler"));set_time_limit(0);ini_set(&＃39;default_socket_timeout&＃39;, -1); //队列处理不超时,解决redis报错:read error on connection}private function redis_client(){$rds &＃61; new Redis();$rds->connect(&＃39;redis.master.host&＃39;,6379);return $rds;}public function process(swoole_process $worker){// 第一个处理$GLOBALS[&＃39;worker&＃39;] &＃61; $worker;swoole_event_add($worker->pipe, function($pipe) {$worker &＃61; $GLOBALS[&＃39;worker&＃39;];$recv &＃61; $worker->read(); //send data to mastersleep(rand(1, 3));echo "From Master: $recvn";$worker->exit(0);});exit;}public function testAction(){for ($i &＃61; 0; $i <10000; $i&＃43;&＃43;){$data &＃61; [&＃39;abc&＃39; &＃61;> $i,&＃39;timestamp&＃39; &＃61;> time().rand(100,999)];$this->masterRedis->lpush($this->redis_task_wing, json_encode($data));}exit;}public function runAction(){while (1){
// echo "t now we de have $this->child child processesn";if ($this->child <$this->maxProcesses){$rds &＃61; $this->redis_client();$data_pop &＃61; $rds->brpop($this->redis_task_wing, 3);//无任务时,阻塞等待if (!$data_pop){continue;}echo "t Starting new child | now we de have $this->child child processesn";$this->child&＃43;&＃43;;$process &＃61; new swoole_process([$this, &＃39;process&＃39;]);$process->write(json_encode($data_pop));$pid &＃61; $process->start();}}}private function sig_handler($signo) {
// echo "Recive: $signo rn";switch ($signo) {case SIGCHLD:while($ret &＃61; swoole_process::wait(false)) {
// echo "PID&＃61;{$ret[&＃39;pid&＃39;]}n";$this->child--;}}}
}
最终&＃xff0c;经过测试&＃xff0c;单核1G的服务器执行1到3秒的任务可以做到800的并发。
如果你喜欢我写的技术文章以及面试总结&＃xff0c;欢迎关注收看我的视频&＃xff0c;并且点赞、收藏、关注我哦。
我是luke&＃xff0c;感谢你的关注&＃xff01;
很多小伙伴在进阶的时候总会遇到一些问题和瓶颈&＃xff0c;业务代码写多了没有方向感&＃xff0c;不知道该从那里入手去提升&＃xff0c;对此我整理了一些资料&＃xff0c;希望能够去帮助到小伙伴们&＃xff0c;可以关注我。并且加入到我的圈子一起学习成长哦【架构师之路】点击链接申请加入圈子
架构师之路 - 知乎www.zhihu.com
京公网安备 11010802041100号