php多进程应用场景实例详解

本文实例讲述了php多进程应用场景。分享给大家供大家参考，具体如下：

pcntl介绍

扩展介绍

php多进程模块依赖pcntl扩展，官方手册介绍：http://php.net/manual/zh/book.pcntl.php

Note:

1. 此扩展在 Windows 平台上不可用。
2. 进程控制不能被应用在Web服务器环境，当其被用于Web服务环境时可能会带来意外的结果。因此，不能再PHP Web开发中使用多进程。

安装扩展

# 通过pecl安装pcntl扩展
sudo pecl install pcntl
# 增加 extension=pcntl.so
sodo vim /etc/php.ini
# 检查扩展是否安装成功
php -m | grep pcntl

　　

处理文件

当一个文件包含许多任务（每个任务一行），并且各任务之间不存在执行的先后顺序关系，可以将文件进行分割（分割后的文件数量与进程数一致），然后使用多进程进行处理。

例如，现在有10个邮箱账号存储在文件mailist.txt中，每次发送邮件需要耗时2s，则采用单进程依次发送完这些邮件需要耗时20。
如果采用多进程，例如3个进程进行处理，首先需要将文件按行数拆分成3个小文件，其中两个文件是4条记录，一个文件是2条记录。每个进程处理一个小文件，则不同进程发送完邮件的耗时为8、8、6，总耗时取最大值为8s。

拆分文件

原始文件 maillist.txt

000000@163.com
111111@163.com
222222@163.com
333333@163.com
444444@163.com
555555@163.com
666666@163.com
777777@163.com
888888@163.com
999999@163.com

拆分操作

split -a 1 -l 4 maillist.txt task

　　

拆分后的文件

taska

000000@163.com
111111@163.com
222222@163.com
333333@163.com

taskb

444444@163.com
555555@163.com
666666@163.com
777777@163.com

taskc

888888@163.com
999999@163.com

相关脚本

多进程调用脚本 text_task.php

$cmds = [
  ['/Users/zhezhao/www/work/text_mail.php','a'],
  ['/Users/zhezhao/www/work/text_mail.php','b'],
  ['/Users/zhezhao/www/work/text_mail.php','c']
];
foreach ($cmds as $cmd){
  $pid = pcntl_fork();
  if($pid == -1){
    exit('create process failed');
  }
  if($pid > 0){
    pcntl_wait($status,WNOHANG);
  }else{
    pcntl_exec('/usr/bin/php',$cmd);
  }
}

　　多进程执行脚本 text_mail.php

require 'MailWork.php';
$name = $argv[1];
echo $name." start #".time().PHP_EOL;
$worker = new MailWork($name);
$res = $worker->text_mail($argc,$argv);
if($res === false){
  echo $worker->getLastError();
}else{
  echo $name." ".$res." works done # ".time().PHP_EOL;
}

　　

输出结果

c start #1504499765
b start #1504499765
a start #1504499765
b#mailto:444444@163.com
c#mailto:888888@163.com
a#mailto:000000@163.com
b#mailto:555555@163.com
a#mailto:111111@163.com
c#mailto:999999@163.com
c 2 works done # 1504499769
a#mailto:222222@163.com
b#mailto:666666@163.com
b#mailto:777777@163.com
a#mailto:333333@163.com
b 4 works done # 1504499773
a 4 works done # 1504499773

在text_task.php中创建了3个进程（a、b、c），其中a和b处理的文件中有4条记录，c处理的文件中有2条记录。

通过输出结果可以得到：

1. a、b、c 三个进程同时开始执行，开始时间戳1504499765
2. c最先完成，完成时间戳1504499769，耗时4s
3. a和c同时完成，完成时间戳1504499773，耗时8s

处理消息队列

这是另外一种常见的常见，我们将耗时操作放入消息队列，通过脚本从消息队列中取出并执行记录。如果通过单个进程依次读取并处理消息，容易使队列中积累大量数据，导致操作的延迟时间较长，这种场景可以通过多个进程来读取并处理消息。redis中的pop操作具有原子性，不会存在多个读取到相同的队列消息的情况。

多进程调用脚本 redis_task.php

$redis = new Redis();
$redis->connect('192.168.1.10');
$task_key = 'task_list';
$task_list = [
  '000000@163.com',
  '111111@163.com',
  '222222@163.com',
  '333333@163.com',
  '444444@163.com',
  '555555@163.com',
  '666666@163.com',
  '777777@163.com',
  '888888@163.com',
  '999999@163.com',
];
foreach ($task_list as $task){
  $redis->lPush($task_key,$task);
}
$cmds = [
  ['/Users/zhezhao/www/work/redis_mail.php','a'],
  ['/Users/zhezhao/www/work/redis_mail.php','b'],
  ['/Users/zhezhao/www/work/redis_mail.php','c']
];
foreach ($cmds as $cmd){
  $pid = pcntl_fork();
  if($pid == -1){
    exit('create process failed');
  }
  if($pid > 0){
    pcntl_wait($status,WNOHANG);
  }else{
    pcntl_exec('/usr/bin/php',$cmd);
  }
}

　　多进程执行脚本 redis_mail.php

require 'MailWork.php';
$name = $argv[1];
echo $name." start #".time().PHP_EOL;
$worker = new MailWork($name);
$redis = new Redis();
$redis->connect('192.168.1.10');
$task_key = 'task_list';
while($redis->lLen($task_key)>0){
  $mailto = $redis->rPop($task_key);
  $worker->redis_mail($mailto);
}
echo $name." work done # ".time().PHP_EOL;

　　

输出结果

b start #1504499844
c start #1504499844
a start #1504499844
b#mailto:000000@163.com
a#mailto:111111@163.com
c#mailto:222222@163.com
b#mailto:333333@163.com
a#mailto:444444@163.com
c#mailto:555555@163.com
b#mailto:666666@163.com
a#mailto:777777@163.com
c#mailto:888888@163.com
c work done # 1504499850
a work done # 1504499850
b#mailto:999999@163.com
b work done # 1504499852

通过输出结果可以得到

1. a、b、c三个进程同时开始执行，时间戳为1504499844
2. a和c同时结束执行，分别处理了3条记录，时间戳为1504499850，耗时6s
3. b最后结束执行，处理了4条记录，时间戳为1504499852，耗时8s

master-worker模式

我们模拟Web服务器处理http请求的操作，对于每个请求创建一个进程，用于处理请求内容。

class WebServer
{
  private $list;
  public function __construct()
  {
    $this->list = [];
  }
  public function worker($request){
    $pid = pcntl_fork();
    if($pid == -1){
      return false;
    }
    if($pid > 0){
      return $pid;
    }
    if($pid == 0){
      $time = $request[0];
      $method = $request[1];
      $start = time();
      echo getmypid()."	 start ".$method."	at".$start.PHP_EOL;
      sleep($time);
      $end = time();
      $cost = $end-$start;
      echo getmypid()."	 stop 	".$method."	at:".$end."	cost:".$cost.PHP_EOL;
      exit(0);
    }
  }
  public function master($requests){
    $start = time();
    echo "All request handle stop at ".$start.PHP_EOL;
    foreach ($requests as $request){
      $pid = $this->worker($request);
      if(!$pid){
        echo 'handle fail!'.PHP_EOL;
        return;
      }
      array_push($this->list,$pid);
    }
    while(count($this->list)>0){
      foreach ($this->list as $k=>$pid){
        $res = pcntl_waitpid($pid,$status,WNOHANG);
        if($res == -1 || $res > 0){
          unset($this->list[$k]);
        }
      }
      usleep(100);
    }
    $end = time();
    $cost = $end - $start;
    echo "All request handle stop at ".$end."	 cost:".$cost.PHP_EOL;
  }
}
$requests = [
 [1,'GET index.php'],
 [2,'GET index.php'],
 [3,'GET index.php'],
 [4,'GET index.php'],
 [5,'GET index.php'],
 [6,'GET index.php']
];
$server = new WebServer();
$server->master($requests);

　　

输出结果：

All request handle stop at 1504513048
18945    start GET index.php    at1504513048
18944    start GET index.php    at1504513048
18946    start GET index.php    at1504513048
18947    start GET index.php    at1504513048
18948    start GET index.php    at1504513048
18949    start GET index.php    at1504513048
18944    stop   GET index.php   at:1504513049   cost:1
18945    stop   GET index.php   at:1504513050   cost:2
18946    stop   GET index.php   at:1504513051   cost:3
18947    stop   GET index.php   at:1504513052   cost:4
18948    stop   GET index.php   at:1504513053   cost:5
18949    stop   GET index.php   at:1504513054   cost:6
All request handle stop at 1504513054    cost:6

如果依次处理请求，总耗时为1+2+3+4+5+6=21s。每个请求创建一个进程的处理方式，总耗时是最耗时的请求操作6s。

多进程最好在方法、函数或者文件中单独使用，这样逻辑更加清晰，也便于分析和维护。

附录

邮件操作类：

MailWork.php

<?php
/**
 * Created by PhpStorm.
 * User: zhezhao
 * Date: 2017/9/4
 * Time: 上午10:05
 */
class MailWork
{
  private $error;
  private $name;
  public function __construct($name)
  {
    $this->name = $name;
  }
  public function getLastError(){
    return $this->error;
  }
  public function checkEnv($argc)
  {
    if (substr(php_sapi_name(), 0, 3) !== 'cli') {
      $this->error ="This Programe can only be run in CLI mode";
      return false;
    }
    if($argc!=2){
      $this->error = 'wrong params';
      return false;
    }
    return true;
  }
  public function getFileName($argv){
    $filename = "task".$argv[1];
    if(!file_exists($filename)){
      $this->error = 'file does not exits';
      return false;
    }else{
      return $filename;
    }
  }
  public function sendMail($mailto)
  {
    sleep(2);
    echo $this->name."#mailto:".$mailto.PHP_EOL;
  }
  public function text_mail($argc,$argv){
    if(!$this->checkEnv($argc)){
      return false;
    }
    $filename = $this->getFileName($argv);
    if(!$filename){
      return false;
    }
    $fp = fopen($filename,'r');
    $counter = 0;
    while(!feof($fp)){
      $line = fgets($fp);
      $mailto = substr($line,0,strlen($line)-1);
      if(preg_match('/^w+@w+.w+$/',$mailto)){
        $this->sendMail($mailto);
        $counter++;
      }
    }
    return $counter;
  }
  public function redis_mail($mailto){
    if(preg_match('/^w+@w+.w+$/',$mailto)){
      $this->sendMail($mailto);
    }
  }
}

　　

 

 

PHP使用PCNTL系列的函数也能做到多进程处理一个事务。比如我需要从数据库中获取80w条的数据，再做一系列后续的处理，这个时候，用单进程？你可以等到明年今天了。所以应该使用pcntl函数了。

假设我想要启动20个进程，将1-80w的数据分成20份来做，主进程等待所有子进程都结束了才退出：

$max = 800000;
$workers = 20;
$pids = array();
for($i = 0; $i < $workers; $i++){
  $pids[$i] = pcntl_fork();
  switch ($pids[$i]) {
    case -1:
      echo "fork error : {$i} 
";
      exit;
    case 0:
      $param = array(
        'lastid' => $max / $workers * $i,
        'maxid' => $max / $workers * ($i+1),
      );
      $this->executeWorker($input, $output, $param);
      exit;
    default:
      break;
  }
}
foreach ($pids as $i => $pid) {
  if($pid) {
    pcntl_waitpid($pid, $status);
  }
}

　　

这里当pcntl_fork出来以后，会返回一个pid值，这个pid在子进程中看是0，在父进程中看是子进程的pid（>0），如果pid为-1说明fork出错了。

使用一个$pids数组就可以让主进程等候所有进程完结之后再结束了