深入解析PHP中的(偽)多線程與多進程

本篇文章是對PHP中的(偽)多線程與多進程進行了詳細的分析介紹，需要的朋友參考下  

(偽)多線程：借助外力
利用WEB服務器本身的多線程來處理，從WEB服務器多次調用我們需要實現多線程的程序。
QUOTE:
我們知道PHP本身是不支持多線程的, 但是我們的WEB服務器是支持多線程的.
也就是說可以同時讓多人一起訪問. 這也是我在PHP中實現多線程的基礎.
假設我們現在運行的是a.php這個文件. 但是我在程序中又請求WEB服務器運行另一個b.php
那麼這兩個文件將是同時執行的.
(PS: 一個鏈接請求發送之後, WEB服務器就會執行它, 而不管客戶端是否已經退出)
有些時候, 我們想運行的不是另一個文件, 而是本文件中的一部分代碼.該怎麼辦呢?
其實可是通過參數來控制a.php來運行哪一段程序.
下面看一個例子:

復制代碼 代碼如下:
<?php
function runThread(){
$fp = fsockopen('localhost', 80, $errno, $errmsg);
fputs($fp, "GET /a.php?act=brnrn");//這裡的第二個參數是HTTP協議中規定的請求頭，不明白的請看RFC中的定義
fclose($fp);
}
function a(){
$fp = fopen('result_a.log', 'w');
fputs($fp, 'Set in ' . Date('h:i:s', time()) . (double)microtime() . "rn");
fclose($fp);
}
function b(){
$fp = fopen('result_b.log', 'w');
fputs($fp, 'Set in ' . Date('h:i:s', time()) . (double)microtime() . "rn");
fclose($fp);
}
if(!isset($_GET['act'])){ $_GET['act'] = 'a';};
if($_GET['act'] == 'a'){
runThread();
a();
}else if($_GET['act'] == 'b'){
b();
};
?>

打開result_a.log 和 result_b.log 比較一下兩個文件的中訪問的時間. 大家會發現, 這兩個的確是在不同線程中運行的.有些時間完全一樣.
上面只是一個簡單的例子, 大家可以改進成其它形式.
既然PHP中也能多線程了, 那麼問題也來了, 那就是同步的問題. 我們知道 PHP本身是不支持多線程的. 所以更不會有什麼像Java 中synchronize的方法了. 那我們該如何做呢.

1. 盡量不訪問同一個資源. 以避免沖突. 但是可以同時像數據庫操作. 因為數據庫是支持並發操作的. 所以在多線程的PHP中
不 要向同一個文件中寫入數據. 如果必須要寫的話, 用別的方法進行同步.. 如調用 flock對文件進行加鎖等. 或建立臨時文件並在另外的線程中等待這個文件的消失 while(file_exits('xxx')); 這樣就等於這個臨時文件存在時, 表示其實線程正在操作,如果沒有了這個文件, 說明其它線程已經釋放了這個.

2. 盡量不要從runThread在執行fputs後取這個socket中讀取數據. 因為要實現多線程, 需要的用非阻塞模式. 即在像fgets這樣的函數時立即返回.. 所以讀寫數據就會出問題. 如果使用阻塞模式的話, 程序就不算是多線程了. 他要等上面的返回才執行下面的程序. 所以如果需要交換數據最後利用外面文件或數據中完成. 實在想要的話就用socket_set_nonblock($fp) 來實現.

說了這麼多, 倒底這個有沒有實際的意義呢? 在什麼時候需要這種用這種方法呢 ?
答案是肯定的. 大家知道. 在一個不斷讀取網絡資源的應用中, 網絡的速度是瓶頸. 如果采多這種形式就可以同時以多個線程對不同的頁面進行讀取.

本人做的一個能從8848、soaso這些商城網站搜索信息的程序。還有一個從阿裡巴巴網站上讀取商業信息和公司目錄的程序也用到了此技術。 因為這兩個程序都是要不斷的鏈接它們的服務器讀取信息並保存到數據庫。 利用此技術正好消除了在等待響應時的瓶頸。

多進程：使用PHP的Process Control Functions(PCNTL/線程控制函數)
只能用在Unix Like OS，Windows不可用。
編譯php的時候，需要加上--enable-pcntl，且推薦僅僅在CLI模式運行，不要在WEB服務器環境運行。
以下為簡短的測試代碼：

復制代碼 代碼如下:
declare(ticks=1);
$bWaitFlag = FALSE; /// 是否等待進程結束
$intNum = 10; /// 進程總數
$pids = array(); /// 進程PID數組
echo ("Startn");
for($i = 0; $i < $intNum; $i++) {
$pids[$i] = pcntl_fork();/// 產生子進程，而且從當前行之下開試運行代碼，而且不繼承父進程的數據信息
if(!$pids[$i]) {
// 子進程進程代碼段_Start
$str="";
sleep(5+$i);
for ($j=0;$j<$i;$j++) {$str.="*";}
echo "$i -> " . time() . " $str n";
exit();
// 子進程進程代碼段_End
}
}
if ($bWaitFlag)
{
for($i = 0; $i < $intNum; $i++) {
pcntl_waitpid($pids[$i], $status, WUNTRACED);
echo "wait $i -> " . time() . "n";
}
}
echo ("Endn");

運行結果如下：
CODE:[Copy toclipboard][qiao@oicq qiao]$ phptest.php
Start
End
[qiao@oicq qiao]$ ps -aux | grep "php"
qiao 32275 0.0 0.5 49668 6148pts/1 S 14:03 0:00/usr/local/php4/b
qiao 32276 0.0 0.5 49668 6152pts/1 S 14:03 0:00/usr/local/php4/b
qiao 32277 0.0 0.5 49668 6152pts/1 S 14:03 0:00/usr/local/php4/b
qiao 32278 0.0 0.5 49668 6152pts/1 S 14:03 0:00/usr/local/php4/b
qiao 32279 0.0 0.5 49668 6152pts/1 S 14:03 0:00/usr/local/php4/b
qiao 32280 0.0 0.5 49668 6152pts/1 S 14:03 0:00 /usr/local/php4/b
qiao 32281 0.0 0.5 49668 6152pts/1 S 14:03 0:00/usr/local/php4/b
qiao 32282 0.0 0.5 49668 6152pts/1 S 14:03 0:00/usr/local/php4/b
qiao 32283 0.0 0.5 49668 6152pts/1 S 14:03 0:00/usr/local/php4/b
qiao 32284 0.0 0.5 49668 6152pts/1 S 14:03 0:00/usr/local/php4/b
qiao 32286 0.0 0.0 1620 600pts/1 S 14:03 0:00 grep php
[qiao@oicq qiao]$ 0 -> 1133503401
1 -> 1133503402 *
2 -> 1133503403 **
3 -> 1133503404 ***
4 -> 1133503405 ****
5 -> 1133503406 *****
6 -> 1133503407 ******
7 -> 1133503408 *******
8 -> 1133503409 ********
9 -> 1133503410 *********
[qiao@oicq qiao]$
如果$bWaitFlag=TURE，則結果如下：
CODE:[Copy toclipboard][qiao@oicq qiao]$ phptest.php
Start
0 -> 1133503602
wait 0 -> 1133503602
1 -> 1133503603 *
wait 1 -> 1133503603
2 -> 1133503604 **
wait 2 -> 1133503604
3 -> 1133503605 ***
wait 3 -> 1133503605
4 -> 1133503606 ****
wait 4 -> 1133503606
5 -> 1133503607 *****
wait 5 -> 1133503607
6 -> 1133503608 ******
wait 6 -> 1133503608
7 -> 1133503609 *******
wait 7 -> 1133503609
8 -> 1133503610 ********
wait 8 -> 1133503610
9 -> 1133503611 *********
wait 9 -> 1133503611
End
[qiao@oicq qiao]$
從 多進程的例子可以看出，使用pcntl_fork()之後，將生成一個子進程，而且子進程運行的代碼，從pcntl_fork()之後的代碼開始，而子進 程不繼承父進程的數據信息(實際上是把父進程的數據做了一個全新的拷貝)，因而使用if(!$pids[$i]) 來控制子進程實際運行的代碼段。
更詳細的研究出於時間關系，暫時沒有進行，你可以參考我給出的手冊的鏈接。

[文章二] 嘗試php命令行腳本多進程並發執行
除了fork, cli下的並發方式還有一種,看我的例子:
php不支持多線程，但是我們可以把問題轉換成“多進程”來解決。由於php中的pcntl_fork只有unix平台才可以使用，所以本文嘗試使用popen來替代。
下面是一個例子：
被並行調用的子程序代碼：

復制代碼 代碼如下:
<?php
if($argc==1){
echo("argvn");
}
$arg = $argv[1];
for($i=0; $i<10; $i++)
{
echo($i.".1.".time()." exec $arg n");
if($arg=='php2'){
sleep(1);
echo($i.".2.".time()." exec $arg n");
sleep(1);
}else{
sleep(1);
}
}
?>

主調用者程序，由他調用子進程，同時並發的收集子程序的輸出

復制代碼 代碼如下:
error_reporting(E_ALL);
$handle1 = popen('php sub.php php1', 'r');
$handle2 = popen('php sub.php php2', 'r');
$handle3 = popen('php sub.php php3', 'r');
echo "'$handle1'; " . gettype($handle1) . "n";
echo "'$handle2'; " . gettype($handle2) . "n";
echo "'$handle3'; " . gettype($handle3) . "n";
//sleep(20);
while(!feof($handle1) || !feof($handle2) || !feof($handle3) )
{
$read = fgets($handle1);
echo $read;
$read = fgets($handle2);
echo $read;
$read = fgets($handle3);
echo $read;
}
pclose($handle1);
pclose($handle2);
pclose($handle3);

下面是我機器上的輸出：
C:my_hunter>php exec.php
'Resource id #4'; resource
'Resource id #5'; resource
'Resource id #6'; resource
0.1.1147935331 exec php1
0.1.1147935331 exec php2
0.1.1147935331 exec php3
1.1.1147935332 exec php1
0.2.1147935332 exec php2
1.1.1147935332 exec php3
2.1.1147935333 exec php1
1.1.1147935333 exec php2
2.1.1147935333 exec php3
3.1.1147935334 exec php1
1.2.1147935334 exec php2
3.1.1147935334 exec php3
4.1.1147935335 exec php1
2.1.1147935335 exec php2
4.1.1147935335 exec php3
5.1.1147935336 exec php1
2.2.1147935336 exec php2
5.1.1147935336 exec php3
6.1.1147935337 exec php1
3.1.1147935337 exec php2
6.1.1147935337 exec php3
7.1.1147935338 exec php1
3.2.1147935338 exec php2
7.1.1147935338 exec php3
8.1.1147935339 exec php1
4.1.1147935339 exec php2
8.1.1147935339 exec php3
9.1.1147935340 exec php1
4.2.1147935340 exec php2
9.1.1147935340 exec php3
5.1.1147935341 exec php2
5.2.1147935342 exec php2
6.1.1147935343 exec php2
6.2.1147935344 exec php2
7.1.1147935345 exec php2
7.2.1147935346 exec php2
8.1.1147935347 exec php2
8.2.1147935348 exec php2
9.1.1147935349 exec php2
9.2.1147935350 exec php2
**總結：**
**主程序循環等待子進程， 通過fgets或fread 把子進程的輸出獲取出來 ， 從時間戳上看，的確實現了並發執行。**
-----------------------------------------------
以後的改進：
* popen打開的句柄是單向的，如果需要向子進程交互，可以使用proc_open
* 使用數組和子函數代替while(!feof($handle1)|| !feof($handle2) || !feof($handle3) )這種龌龊的寫法
* 用fread一次把子進程已經產生的輸出取完，而不是每次一行。
一個並發執行shell任務的調度者,本程序讀取一個任務文件,把裡面的每行命令並發執行, 可以設置同時存在的子進程數目:

復制代碼 代碼如下:
/*
主任務管理器
並發的執行子任務列表
*/
include("../common/conf.php");
include("../common/function.php");
//開啟的進程數
$exec_number = 40 ;
/***** main ********/
if($argc==1){
echo("argvn");
}
$taskfile = $argv[1];
//tasklist
$tasklist = file($taskfile);
$tasklist_len = count($tasklist);
$tasklist_pos = 0;
$handle_list = array();
while(1)
{
//子進程列表有空閒,則填充補齊子進程列表
if($exec_number > count($handle_list) &&
$tasklist_pos < $tasklist_len)
{
for($i=$tasklist_pos; $i<$tasklist_len; )
{
$command = $tasklist[$i] ;
$handle_list[] = popen($command , "r" );
tolog("begin task t ".$tasklist[$i]);
$i++;
if($exec_number == count($handle_list)) break;
}
$tasklist_pos = $i;
}
//如果子進程列表空，退出
if(0 == count($handle_list))
{
break;
}
//檢查子進程列表的輸出，把停掉的子進程關閉並記錄下來
$end_handle_keys = array();
foreach($handle_list as $key => $handle)
{
//$str = fgets($handle, 65536);
$str = fread($handle, 65536);
echo($str);
if(feof($handle))
{
$end_handle_keys[] = $key;
pclose($handle);
}
}
//踢出停掉的子進程
foreach($end_handle_keys as $key)
{
unset($handle_list[$key]);
//var_dump($handle_list);
//exit;
}
}
tolog("nn*******************end**********************nn", "" , true);

附加一段Socket多進程接收的代碼：

復制代碼 代碼如下:

do {
if (($msgsock = socket_accept($sock)) < 0) {
echo "socket_accept() failed: reason: " . socket_strerror($msgsock) . "n";
break;
}
$pid = pcntl_fork();
if ($pid == -1) {
die('could not fork');
} else if (!$pid) {
.....
socket_write($msgsock, $msg, strlen($msg));
do {
......
} while (true);
socket_close($msgsock);
}
} while (true);