本页文章导读:

    ▪php实现多进程 多任务的一例代码      代码如下：   代码示例: <?php /** *基于PHP5实现 *借助proc_open *能启动多进程，你可以使用你的想象力做你想做的了 *如果你是在linux上跑php，并且启用pcntl模块后，使用pcntl函数该更好 *.........

    ▪php进程锁定相关问题的实例剖析      本文要学到的有关php进程的知识点： 1. 区分读锁定 和 写 锁定。 如果每次都使用 写锁定，那么连多个进程读取一个文件也要排队，这样的效率肯定不行。 2. 区分 阻塞 与 非 阻塞模式。 一.........

    ▪php 多进程编程的深入探讨      使用PHP真正的多进程运行模式，适用于数据采集、邮件群发、数据源更新、tcp服务器等环节。 PHP有一组进程控制函数(编译时需要 –enable-pcntl与posix扩展)，使得php能在*nix系统中实现跟c一样的.........

代码示例:

<?php
/**
*基于PHP5实现
*借助proc_open
*能启动多进程，你可以使用你的想象力做你想做的了
*如果你是在linux上跑php，并且启用pcntl模块后，使用pcntl函数该更好
*最后修改：by www. 2013/6/20
**/
    error_reporting(E_ALL); 
    set_time_limit(0); 

    class Thread { 
    protected $_pref; // process reference 
    protected static $_instance = null; 
    protected $_pipes; 

    private function __construct() { 
    $this->_pref = 0; 
    } 

    public static function getInstance($file) { 
    if (null == self::$_instance) { 
    self::$_instance = new self; 
    } 

    $descriptor = array( 
    0 => array("pipe", "r"), 
    1 => array("pipe", "w"), 
    2 => array("file", "./error-output.txt", "a"), 
    ); 
    self::$_instance->_pref = proc_open("php -q $file", $descriptor, self::$_instance->_pipes); 
    return true; 
    } 

    public function __destruct() { 
    proc_close($this->_pref); 
    $this->_pref = null; 
}
?>

1. 区分读锁定 和 写 锁定。
如果每次都使用 写锁定，那么连多个进程读取一个文件也要排队，这样的效率肯定不行。

2. 区分 阻塞 与 非 阻塞模式。
一般来说，如果一个进程在写一个文件的时候，另外一个进程应该被阻塞，但很多时候，可以先干点别的事情，然后再判断一下是否有其他人在写文件，如果没有，再加入数据，这样的效率更高。

3. 修复了 锁定文件在linux 上的bug，特别是 在 gfs 文件系统上的bug。

代码示例:

<?php
/**
* php 进程锁相关问题
* edit www.
*/
class File_Lock
{
private $name;
private $handle;
private $mode;
function __construct($filename, $mode = 'a+b')
{
global $php_errormsg;
$this->name = $filename;
$path = dirname($this->name);
if ($path == '.' || !is_dir($path)) {
global $config_file_lock_path;
$this->name = str_replace()(array("/", "\\"), array("_", "_"), $this->name);
if ($config_file_lock_path == null) {
$this->name = dirname(__FILE__) . "/lock/" . $this->name;
} else {
$this->name = $config_file_lock_path . "/" . $this->name;
}
}
$this->mode = $mode;
$this->handle = @fopen($this->name, $mode);
if ($this->handle == false) {
throw new Exception($php_errormsg);
}
}
public function close()
{
if ($this->handle !== null ) {
@fclose($this->handle);
$this->handle = null;
}
}
public function __destruct()
{
$this->close();
}
public function lock($lockType, $nonBlockingLock = false)
{
if ($nonBlockingLock) {
return flock($this->handle, $lockType | LOCK_NB);
} else {
return flock($this->handle, $lockType);
}
}
public function readLock()
{
return $this->lock(LOCK_SH);
}
public function writeLock($wait = 0.1)
{
$startTime = microtime(true);
$canWrite = false;
do {
$canWrite = flock($this->handle, LOCK_EX);
if(!$canWrite) {
usleep(rand(10, 1000));
}
} while ((!$canWrite) && ((microtime(true) - $startTime) < $wait));
}
/**
* if you want't to log the number under multi-thread system,
* please open the lock, use a+ mod. then fopen the file will not
* destroy the data.
*
* this function increment a delt value , and save to the file.
*
* @param int $delt
* @return int
*/
public function increment($delt = 1)
{
$n = $this->get();
$n += $delt;
$this->set($n);
return $n;
}
public function get()
{
fseek($this->handle, 0);
return (int)fgets($this->handle);
}
public function set($value)
{
ftruncate($this->handle, 0);
return fwrite($this->handle, (string)$value);
}
public function unlock()
{
if ($this->handle !== null ) {
return flock($this->handle, LOCK_UN);
} else {
return true;
}
}
}
?>

代码示例:

<?php
/**
* 进行写锁定的测试
* 打开线程1
*/
require("file_lock.php");
$lock = new File_Lock(dirname(dirname(__FILE__)) . "/FileLock.lock");
/** 单个线程锁定的速度 1s 钟 3万次。 **/
/** 两个线程写，两万的数据 大概要 7s 钟*/
/** 一个线程写，一万的数据 大概要 3.9s 钟，居然两个文件同时写，要快一点*/
/** 不进行锁定，一个进程 写大概要 2.8s 钟，加锁是有代价的。 */
/** 不进行锁定，两个进程 分布不是很均匀，而且大多数都冲突 */
$lock->writeLock();
$lock->increment();
$lock->unlock();
while ($lock->get() < 2) {
usleep(1000);
}
sleep(1);
echo "begin to runing \n";
$t1 = microtime(true);
for ($i = 0; $i < 10000; $i++)
{
$lock->writeLock();
$lock->increment(1);
$lock->unlock();
}
$t2 = microtime(true) - $t1;
echo $t2;
?>

代码示例:

<?php
declare(ticks = 1); 

        function signal_handler($signal) { 
            print "Caught SIGALRM\n"; 
            pcntl_alarm(5); 
        } 

        pcntl_signal(SIGALRM, "signal_handler", true); 
        pcntl_alarm(5); 

        for(;;) { 
        } 
?>
 

代码示例:

<?php
$dir = '/home/shankka/'; 
    $cmd = 'ls'; 
    $option = '-l'; 
    $pathtobin = '/bin/ls'; 

    $arg = array($cmd, $option, $dir); 

    pcntl_exec($pathtobin, $arg); 
    echo '123';    //不会执行到该行 
?>

代码示例:

<?php
$pid = pcntl_fork();
    //这里最好不要有其他的语句
    if ($pid == -1) {
      die('could not fork');
    } else if ($pid) {
      // we are the parent
      pcntl_wait($status); //Protect against Zombie children
    } else {
      // we are the child
    }
?>

代码示例:

<?php
$pid = pcntl_fork(); 
    if($pid) { 
       pcntl_wait($status); 
       $id = getmypid(); 
       echo "parent process,pid {$id}, child pid {$pid}\n"; 
    }else{ 
       $id = getmypid(); 
       echo "child process,pid {$id}\n"; 
       sleep(2); 
    } 
?>

代码示例:

<?php
/**
* Project: Signfork: php多线程库
* File:    Signfork.class.php
* modify by www.
*/

class Signfork{
  /**
   * 设置子进程通信文件所在目录
   * @var string
   */
  private $tmp_path='/tmp/';

/**
  * Signfork引擎主启动方法
  * 1、判断$arg类型,类型为数组时将值传递给每个子进程;类型为数值型时,代表要创建的进程数.
  * @param object $obj 执行对象
  * @param string|array $arg 用于对象中的__fork方法所执行的参数
  * 如:$arg,自动分解为:$obj->__fork($arg[0])、$obj->__fork($arg[1])...
  * @return array  返回   array(子进程序列=>子进程执行结果);
  */
  public function run($obj,$arg=1){
    if(!method_exists($obj,'__fork')){
      exit("Method '__fork' not found!");
    }

    if(is_array($arg)){
     $i=0;
     foreach($arg as $key=>$val){
       $spawns[$i]=$key;
       $i++;
       $this->spawn($obj,$key,$val);
     }
     $spawns['total']=$i;
    }elseif($spawns=intval($arg)){
      for($i = 0; $i < $spawns; $i++){
        $this->spawn($obj,$i);
      }
    }else{
      exit('Bad argument!');
    }

   if($i>1000) exit('Too many spawns!');
      return $this->request($spawns);
   }

  /**
   * Signfork主进程控制方法
   * 1、$tmpfile 判断子进程文件是否存在，存在则子进程执行完毕，并读取内容
   * 2、$data收集子进程运行结果及数据，并用于最终返回
   * 3、删除子进程文件
   * 4、轮询一次0.03秒，直到所有子进程执行完毕，清理子进程资源
   * @param  string|array $arg 用于对应每个子进程的ID
   * @return array  返回   array([子进程序列]=>[子进程执行结果]);
   */
   private function request($spawns){
     $data=array();
     $i=is_array($spawns)?$spawns['total']:$spawns;
     for($ids = 0; $ids<$i; $ids++){
       while(!($cid=pcntl_waitpid(-1, $status, WNOHANG)))usleep(30000);
       $tmpfile=$this->tmp_path.'sfpid_'.$cid;
       $data[$spawns['total']?$spawns[$ids]:$ids]=file_get_contents($tmpfile);
       unlink($tmpfile);
     }
     return $data;
   }

/**
  * Signfork子进程执行方法
  * 1、pcntl_fork 生成子进程
  * 2、file_put_contents 将'$obj->__fork($val)'的执行结果存入特定序列命名的文本
  * 3、posix_kill杀死当前进程
  * @param object $obj        待执行的对象
  * @param object $i                子进程的序列ID，以便于返回对应每个子进程数据
  * @param object $param 用于输入对象$obj方法'__fork'执行参数
  */
  private function spawn($obj,$i,$param=null){
    if(pcntl_fork()===0){
      $cid=getmypid();
      file_put_contents($this->tmp_path.'sfpid_'.$cid,$obj->__fork($param));
      posix_kill($cid, SIGTERM);
      exit;
    }
  }
}
?>