php-fpm，以及现在比较火的workman，swoole，其都是使用的多进程的管理模式，使用manger进程，管理其子进程，今天通过纯php做了一个类似的小东西，这样以后再做守护脚本的时候，可以使用这种方式，好处就是，多进程，更能利用多核的优势，并且可以做平滑重启。

本代码实现的功能，可以自定义子进程数，比如你定3个子进程，运行会产生1个manger进程，3个work进程，通过linux信号，来实现，进程的平滑重启。

运行效果：

  1. [zhangcunchao@localhost evente-member-client]$ php7  bin/timer/console.php queue
  2. 0
  3. 2017-09-08 17:59:54 work proccess start
  4. 2017-09-08 17:59:54 work proccess start
  5. 2017-09-08 17:59:54 work proccess start
  6. 2017-09-08 17:59:54 work proccess start
  7. 2017-09-08 17:59:54 work proccess start
  1. [zhangcunchao@localhost evente-member-service]$ ps -ef|grep queue
  2. 541      15340  7098  0 18:30 pts/14   00:00:00 development-member-queue-manger                           
  3. 541      15377 15340  0 18:30 pts/14   00:00:00 development-member-queue-work                             
  4. 541      15378 15340  0 18:30 pts/14   00:00:00 development-member-queue-work                             
  5. 541      15379 15340  0 18:30 pts/14   00:00:00 development-member-queue-work                             
  6. 541      15380 15340  0 18:30 pts/14   00:00:00 development-member-queue-work                             
  7. 541      15381 15340  0 18:30 pts/14   00:00:00 development-member-queue-work                             
  8. 541      15863  6992  0 18:30 pts/9    00:00:00 grep --color=auto queue

主要代码

使用了Symfony 做cli任务

  1. <?php
  3. namespace App\Console\Timer;
  5. use Symfony\Component\Console\Command\Command;
  6. use Symfony\Component\Console\Input\InputInterface;
  7. use Symfony\Component\Console\Output\OutputInterface;
  8. use Symfony\Component\Console\Input\InputArgument;
  11. use App\Console\Timer\QueueTools;
  14. /**
  15.  * 多进程守护队列服务
  16.  */
  17. class QueueCmd extends Command
  18. {
  20.     const      proccess_num  = 5;//work子进程数
  21.     protected  $proccess_name = 'member-queue';//进程名前缀，此设置主要为监控脚本提供方便
  22.     protected  $open_debug   = false;//开启debug,警告直接输出，不发邮件
  23.     protected  $_pid;
  24.     protected  $shm_key;      //共享内存key     `
  25.     protected  $shm_id;
  26.     protected  $max_job_limit= 1000; //单个子进程最多处理的任务数，达到此任务就重启此子进程;0 没有限制
  27.     protected  $NOW_DIR;
  28.     protected  $max_try_num   = 3;//队列数据最大尝试次数//超过计入日志
  29.     protected  $time_out      = 20;//子进程任务获取单次循环最大等待时间
  30.     protected  $email_title   = 'php-timer告警';
  32.     public function __construct()
  33.     {
  34.         $this->NOW_DIR =  dirname(__FILE__).'/';
  35.         $this->proccess_name = APP_ENV.'-'.$this->proccess_name;
  36.         parent::__construct();
  37.     }
  39.     protected function configure()
  40.     {
  41.         $this->setName('queue');
  43.     }
  45.     protected function execute(InputInterface $input, OutputInterface $output)
  46.     {
  47.         error_reporting(E_ALL & ~E_NOTICE);
  48.         //检查扩展
  49.         if(!extension_loaded('pcntl')){
  50.             $output->writeln("<error>需要安装pcntl扩展支持</error>");
  51.             exit();
  52.         }
  56.         $shell = "ps w -C php|grep '{$this->proccess_name}-manger'|wc -l";
  57.         $msg = system($shell);
  58.         if($msg > 0)
  59.         {
  60.             $output->writeln("<error>{$this->proccess_name}-manger 任务已经存在</error>");
  61.             exit();
  62.         }
  65.         if(self::proccess_num > 0){
  67.             //启动进程
  68.             for ($i=0; $i<self::proccess_num; $i++)
  69.             {
  70.                 $this->parentProccess();
  72.             }
  73.             pcntl_signal(SIGUSR1, function(){
  74.                 $this->sig_handler(SIGUSR1);
  75.             });
  76.             // 程序结束(terminate)信号, 与SIGKILL不同的是该信号可以被阻塞和 处理. 通常用来要求程序自己正常退出.  shell命令kill缺省产生这 个信号.
  77.             pcntl_signal(SIGTERM, function(){
  78.                 $this->sig_handler(SIGTERM);
  79.             });
  80.             //终止进程  终端线路挂断
  81.             pcntl_signal(SIGHUP, function(){
  82.                 $this->sig_handler(SIGHUP);
  83.             });
  84.             cli_set_process_title($this->proccess_name . '-manger');
  87.             while (true) {
  88.                 // do something
  89.                 $pid = pcntl_waitpid(-1, $status, WUNTRACED | WNOHANG);
  90.                 if($pid>0){
  91.                     $s = pcntl_wifexited($status);
  92.                     //todo非正常退出报警
  93.                     $this->parentProccess($pid);
  94.                 }
  96.                 pcntl_signal_dispatch(); // 接收到信号时，调用注册的signalHandler()
  97.                 usleep(100000);
  98.             }
  100.         }else{
  101.             $output->writeln("<error>proccess_num must > 0</error>");
  102.         }
  106.     }
  109.     function sig_handler($signo){
  110.         switch ($signo) {
  111.             case SIGUSR1:
  112.                 foreach($this->_pid as $key){
  113.                     posix_kill($key,SIGUSR1);
  114.                 }
  115.                 echo date('Y-m-d H:i:s')." [manage] [SIGUSR1] all work reload\n";
  116.                 break;
  117.             case SIGTERM:
  118.                 foreach($this->_pid as $key){
  119.                     posix_kill($key,SIGUSR1);
  120.                 }
  121.                 echo date('Y-m-d H:i:s')." [manage] [SIGTERM] stop\n";
  122.                 die();
  123.                 break;
  124.             case SIGHUP:
  125.                 echo date('Y-m-d H:i:s')." [manage] [SIGHUP] linux clinet exit\n";
  126.                 break;
  127.         }
  128.     }
  130.     function sig_san_handler($signo){
  131.         switch ($signo) {
  132.             case SIGUSR1:
  133.                 echo date('Y-m-d H:i:s')." [work] [SIGUSR1] work die\n";
  134.                 die();
  135.                 break;
  136.             case SIGTERM:
  137.                 echo date('Y-m-d H:i:s')." [work] [SIGTERM] work die\n";
  138.                 die();
  139.                 break;
  140.             case SIGHUP:
  141.                 echo date('Y-m-d H:i:s')." [work] [SIGHUP] linux clinet exit\n";
  142.                 break;
  143.         }
  144.     }
  146.     //父进程
  147.     public function parentProccess($id=''){
  148.         if(!empty($id)){
  149.             unset($this->_pid[$id]);
  150.         }
  151.         $pid  = pcntl_fork();
  152.         if($pid == -1)
  153.         {
  154.             return false;
  155.         }
  156.         else
  157.         {
  158.             if($pid)
  159.             {
  160.                 #父进程获得子进程的pid，存入数组
  161.                 $this->_pid[$pid] = $pid;
  162.                 return true;
  163.             }
  164.             else
  165.             {
  166.                 //开始发送，子进程执行完自己的任务后，退出。
  167.                 $this->sanProccess();
  168.                 exit;
  169.             }
  170.         }
  171.     }
  173.     //子进程
  174.     public function sanProccess(){
  176.         //信号绑定
  177.         cli_set_process_title($this->proccess_name . '-work');
  178.         echo date('Y-m-d H:i:s')." work proccess start\n";
  179.         pcntl_signal(SIGUSR1, function(){
  180.             $this->sig_san_handler(SIGUSR1);
  181.         });
  182.         pcntl_signal(SIGHUP, function(){
  183.             $this->sig_san_handler(SIGHUP);
  184.         });
  185.         pcntl_signal(SIGTERM, function(){
  186.             $this->sig_san_handler(SIGTERM);
  187.         });
  190.         $san_job_i = 0;
  193.         while(1){
  194.             //初始容器，避免循环影响
  195.             $job_data = array();
  196.             $job_key  = '';
  197.             $jobs     = [];
  198.             try{
  199.                 //echo '1'.PHP_EOL;//用于排查进程是否异步
  200.                 $jobs = QueueTools::pull_queue($this->time_out);
  201.                 if(!empty($jobs) && !empty($jobs['key']) && !empty($jobs['data'])){
  202.                     $job_data = $jobs['data'];
  203.                     $job_key  = $jobs['key'];
  205.                     switch($job_key){
  207.                         case QueueTools::MEMBER_QUEUE_LIST['member_open'] :
  209.                             print_r($job_data);
  211.                             break;
  213.                         default:
  214.                             app()->log->addError('未定义的队列事件',$jobs);
  216.                     }
  218.                 }
  219.                 //print_r($this->cache->redis->get('wapwx_cookie'));
  221.             }catch (\Exception $e){
  222.                 //统一重试，失败记录重试队列
  223.                 if(!empty($job_data)&&!empty($job_key)){
  224.                     if(isset($job_data['_job_error_num'])){
  225.                         $job_data['_job_error_num']=$job_data['_job_error_num']+1;
  226.                     }else{
  227.                         $job_data['_job_error_num'] = 1;
  228.                     }
  229.                 }
  231.                 //重试上限
  232.                 if($this->max_try_num <= $job_data['_job_error_num']){
  233.                     $mes = '本数据第'.$job_data['_job_error_num'].'次错误尝试'.PHP_EOL;
  234.                     $mes .= '数据已达处理次数上限，重试结束，记入日志'.PHP_EOL;
  236.                     unset($job_data['_job_error_num']);
  238.                     if($this->open_debug){
  239.                         print_r($mes);
  240.                     }else{
  241.                         app()->log->addError($this->email_title,[$mes,$jobs,$e->getCode(),$e->getMessage()]);
  242.                     }
  244.                 }else{
  245.                     //统一重入队列
  246.                     if(!empty($job_data)&&!empty($job_key)){
  248.                         QueueTools::push_queue($job_key,$job_data);
  250.                     }
  252.                 }
  254.             }
  255. //            sleep(3);
  258.             //控制单个进程最大任务数
  259.             if($this->max_job_limit){
  260.                 ++$san_job_i;
  261.                 //echo PHP_EOL;
  262.                 if($this->max_job_limit <= $san_job_i){
  263.                     echo date('Y-m-d H:i:s').' work job limit reload'.PHP_EOL;
  264.                     exit();
  265.                 }
  266.             }
  267.             pcntl_signal_dispatch();
  268.         }
  269.     }
  271. }

App\Console\Timer\QueueTools 代码

  1. <?php
  3. namespace App\Console\Timer;
  5. use App\Redis\Config\RedisKey;
  7. /**
  8.  * 队列服务辅助工具
  9.  */
  10. class QueueTools
  11. {
  12.     const MEMBER_QUEUE_LIST = [
  13.         'member_open'=>'member_open', //会员激活任务
  14.     ];
  17.     //拉取队列任务
  18.     static  function pull_queue($time_out=20)
  19.     {
  20.         $jobs = app()->redis->blpop(RedisKey::MEMBER_QUEUE_KEY,$time_out);
  21.         if(!empty($jobs) && !empty($jobs[1])){
  22.             $data = json_decode($jobs[1],true);
  23.             if(isset($data['key']) && !empty($data['key']) && isset($data['data']) && !empty($data['data'])){
  24.                 return $data;
  25.             }else{
  26.                 return [];
  27.             }
  29.         }else{
  30.             return [];
  31.         }
  33.     }
  35.     //push队列任务
  36.     static  function push_queue($queueKey='',$data = [])
  37.     {
  38.         if(empty($data) || !is_array($data)){
  39.             throw new \Exception('队列任务必须数组',10002);
  40.             return false;
  41.         }
  42.         if(in_array($queueKey,self::MEMBER_QUEUE_LIST)){
  43.             $data = [
  44.                 'key'  => $queueKey,
  45.                 'data' => $data,
  46.             ];
  47.             return app()->redis->rpush(RedisKey::MEMBER_QUEUE_KEY,json_encode($data));
  48.         }
  49.         return false;
  50.     }
  52.     //会员激活任务
  53.     static function open_member_job($orgId=0,$userId=0)
  54.     {
  55.         if(empty($orgId) || empty($userId)){
  56.             return false;
  57.         }
  59.         $data = [
  60.             'orgId'  => $orgId,
  61.             'userId' => $userId,
  62.         ];
  63.         return self::push_queue(self::MEMBER_QUEUE_LIST['member_open'],$data);
  64.     }
  67. }