[原创]PHP多进程运行利器-signfork v1.0.0

bs

使用PHP真正的多进程运行模式(真正利用多核CPU)，适用于数据采集、邮件群发、数据源更新、tcp服务器等环节。
下载地址 http://code.google.com/p/signfork/downloads/list

运行条件：*NUX环境、 编译时需要 --enable-pcntl    posix扩展


以下是主程序文件：

1. 
   
2. <?php
   
3. /**
   
4. * Project: Signfork: php多线程库
   
5. * File:    Signfork.class.php
   
6. *
   
7. * @link        [url]http://code.google.com/p/signfork/[/url]
   
8. * @author        lajabs <hittyo at gmail dot com> QQ:124321697
   
9. * @version 1.0.0 2009/8/4
   
10. */
    
11. 
    
12. 
    
13. 
    
14. class Signfork
    
15. {
    
16.         /**
    
17.      * 设置子进程通信文件所在目录
    
18.      * @var string
    
19.      */
    
20.         private $tmp_path='/tmp/';
    
21. 
    
22.         /**
    
23.      * Signfork引擎主启动方法
    
24.          * 1、判断$arg类型,类型为数组时将值传递给每个子进程;类型为数值型时,代表要创建的进程数.
    
25.          * @param object $obj 执行对象
    
26.          * @param string|array $arg 用于对象中的__fork方法所执行的参数
    
27.          * 如:$arg,自动分解为:$obj->__fork($arg[0])、$obj->__fork($arg[1])...
    
28.          * @return array  返回   array(子进程序列=>子进程执行结果);
    
29.      */
    
30.         public function run($obj,$arg=1)
    
31.         {
    
32.                 if(!method_exists($obj,'__fork'))
    
33.                 {
    
34.                         exit("Method '__fork' not found!");
    
35.                 }
    
36. 
    
37.                 if(is_array($arg))
    
38.                 {
    
39.                         $i=0;
    
40.                         foreach($arg as $key=>$val)
    
41.                         {
    
42.                                 $spawns[$i]=$key;
    
43.                                 $i++;
    
44.                                 $this->spawn($obj,$key,$val);
    
45.                         }
    
46.                         $spawns['total']=$i;
    
47.                 }
    
48.                 elseif($spawns=intval($arg))
    
49.                 {
    
50.                         for($i = 0; $i < $spawns; $i++)
    
51.                         {
    
52.                                 $this->spawn($obj,$i);
    
53.                         }
    
54.                 }
    
55.                 else
    
56.                 {
    
57.                         exit('Bad argument!');
    
58.                 }
    
59.                 if($i>1000) exit('Too many spawns!');
    
60. 
    
61.                 return $this->request($spawns);
    
62.         }
    
63. 
    
64. 
    
65.         /**
    
66.      * Signfork主进程控制方法
    
67.          * 1、$tmpfile 判断子进程文件是否存在，存在则子进程执行完毕，并读取内容
    
68.          * 2、$data收集子进程运行结果及数据，并用于最终返回
    
69.          * 3、删除子进程文件
    
70.          * 4、轮询一次0.03秒，直到所有子进程执行完毕，清理子进程资源
    
71.          * @param  string|array $arg 用于对应每个子进程的ID
    
72.          * @return array  返回   array([子进程序列]=>[子进程执行结果]);
    
73.      */
    
74.         private function request($spawns)
    
75.         {
    
76.                 $data=array();
    
77.                 $i=is_array($spawns)?$spawns['total']:$spawns;
    
78.                 for($ids = 0; $ids<$i; $ids++)
    
79.                 {
    
80.                         while(!($cid=pcntl_waitpid(-1, $status, WNOHANG)))usleep(30000);
    
81.                         $tmpfile=$this->tmp_path.'sfpid_'.$cid;
    
82.                         $data[$spawns['total']?$spawns[$ids]:$ids]=file_get_contents($tmpfile);
    
83.                         unlink($tmpfile);
    
84.                 }
    
85.                 return $data;
    
86.         }
    
87. 
    
88.         /**
    
89.      * Signfork子进程执行方法
    
90.          * 1、pcntl_fork 生成子进程
    
91.          * 2、file_put_contents 将'$obj->__fork($val)'的执行结果存入特定序列命名的文本
    
92.          * 3、posix_kill杀死当前进程
    
93.          * @param object $obj        待执行的对象
    
94.          * @param object $i                子进程的序列ID，以便于返回对应每个子进程数据
    
95.          * @param object $param 用于输入对象$obj方法'__fork'执行参数
    
96.      */
    
97.         private function spawn($obj,$i,$param=null)
    
98.         {
    
99.                 if(pcntl_fork()===0)
    
100.                 {
     
101.                         $cid=getmypid();
     
102.                         file_put_contents($this->tmp_path.'sfpid_'.$cid,$obj->__fork($param));
     
103.                         posix_kill($cid, SIGTERM);
     
104.                         exit;
     
105.                 }
     
106.         }
     
107. }
     
108. 
     
109. ?>
     

复制代码

[ 本帖最后由 bs 于 2009-8-26 01:45 编辑 ]

bs

先做一个类似采集的多进程例子：

1. 
   
2. <?php
   
3. include('Signfork.class.php');
   
4. 
   
5. class test
   
6. {
   
7.         function __fork($arg)
   
8.         {
   
9.                 return file_get_contents($arg);
   
10.         }
    
11. }
    
12. 
    
13. $limit                =microtime(true);
    
14. $test                =new test();
    
15. $Signfork        =new Signfork();
    
16. 
    
17. 
    
18. 
    
19. $arg=array(
    
20. 'http://yahoo.com',
    
21. 'http://baidu.com',
    
22. 'http://google.com',
    
23. 'http://qq.com',
    
24. 'http://163.com',
    
25. 'http://sina.com'
    
26. );
    
27. 
    
28. 
    
29. $Signfork->run($test,$arg);
    
30. echo 'Run time:'.(microtime(true)-$limit);
    
31. ?>
    
32. 
    
33. 输出：
    
34. Run time:1.3541439928133
    
35. 
    
36. 整体运行时间在一秒左右，说明多进程发挥了重要作用。
    

复制代码

[ 本帖最后由 bs 于 2009-8-6 17:43 编辑 ]

bs

php在pcntl_fork()后生成的子进程(通常为僵尸进程)必须由pcntl_waitpid()函数进行资源释放。

注意在pcntl_waitpid()不一定释放的就是当前运行的进程，也可以是过去生成的僵尸进程(没有释放)；也可以是并发时其它访问者的僵尸进程；




PS：僵尸进程

运行
#ps fax|grep defunct

<defunct>

15993 ? Z 0:00 | \ php-cgi?

<defunct>

15791 ? Z 0:00 | \ php-cgi?

<defunct>

15792 ? Z 0:00 | \ php-cgi?

<defunct>

15098 ? Z 0:00 | \ php-cgi?

<defunct>

15921 ? Z 0:00 | \ php-cgi?

<defunct>

....

在进程后面状态为：<defunct>

网鬼

bs是我的偶像

dz902

这是 LZ 写的？

bs

阻断模式运行：
子进程会把运行结果写入进程文件，供主进程获取。
主进程通过pcntl_waitpid得知子进程运行运行结束，收集所有子进程的执行结果并返回。

非阻断模式运行：
主进程仅负责生成子进程和传递参数，不可知且不等待子进程运行结果，运行效率相对阻断模式要高出许多。

xiongchuan86

一定要学习下，都说PHP做采集不好，有了这个，不知道怎么样了，期待啊

hightman

多进程建议在 cli/cgi 方式下运作, 如果是 fastcgi 或 apache 的 isapi module 不太清楚上面的代码是不是会出现一些其它的问题.

而且你这样的处理方式, 最终主程序还是要等所有的子进程完成才能获得返回值; 再有一点主进程可以透过捕捉处理 sigchld 而不必硬生生的 while(waitpid(-1, ... ))

在cli模式下, 可以用 proc_open 及 popen 打开新的进程, 然后主进程中用 stream_select() 来管理它们, 还能完成主进程与子进程的一些交互.

bs

原帖由 hightman 于 2009-8-7 16:54 发表
多进程建议在 cli/cgi 方式下运作, 如果是 fastcgi 或 apache 的 isapi module 不太清楚上面的代码是不是会出现一些其它的问题.

而且你这样的处理方式, 最终主程序还是要等所有的子进程完成才能获得返回值;  ...

嗯，谢谢，之前摸索得比较痛苦，出现过子进程无法即时释放的情况，压力测试一上就挂了。
在子进程并行运行时，主程序理论上只等待一个子进程运行时间，一般还是可以接受的。


目前已经在fastcgi(当前nginx+php-fpm)调试正常，cli模式下我再试试你的方法，再次感谢。