|
|
(偽)多線程：借助外力
利用WEB服務(wù)器本身的多線程來(lái)處理，從WEB服務(wù)器多次調(diào)用我們需要實(shí)現(xiàn)多線程的程序。
QUOTE:
我們知道php本身是不支持多線程的, 但是我們的WEB服務(wù)器是支持多線程的.
也就是說(shuō)可以同時(shí)讓多人一起訪問(wèn). 這也是我在php中實(shí)現(xiàn)多線程的基礎(chǔ).
假設(shè)我們現(xiàn)在運(yùn)行的是a.php這個(gè)文件. 但是我在程序中又請(qǐng)求WEB服務(wù)器運(yùn)行另一個(gè)b.php
那么這兩個(gè)文件將是同時(shí)執(zhí)行的.
(PS: 一個(gè)鏈接請(qǐng)求發(fā)送之后, WEB服務(wù)器就會(huì)執(zhí)行它, 而不管客戶端是否已經(jīng)退出)
有些時(shí)候, 我們想運(yùn)行的不是另一個(gè)文件, 而是本文件中的一部分代碼.該怎么辦呢?
其實(shí)可是通過(guò)參數(shù)來(lái)控制a.php來(lái)運(yùn)行哪一段程序.
下面看一個(gè)例子:
復(fù)制代碼 代碼如下:
<?php
function runThread(){
$fp = fsockopen('localhost', 80, $errno, $errmsg);
fputs($fp, "GET /a.php?act=brnrn");//這里的第二個(gè)參數(shù)是HTTP協(xié)議中規(guī)定的請(qǐng)求頭，不明白的請(qǐng)看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();
};
?>
打開(kāi)result_a.log 和 result_b.log 比較一下兩個(gè)文件的中訪問(wèn)的時(shí)間. 大家會(huì)發(fā)現(xiàn), 這兩個(gè)的確是在不同線程中運(yùn)行的.有些時(shí)間完全一樣.
上面只是一個(gè)簡(jiǎn)單的例子, 大家可以改進(jìn)成其它形式.
既然php中也能多線程了, 那么問(wèn)題也來(lái)了, 那就是同步的問(wèn)題. 我們知道 php本身是不支持多線程的. 所以更不會(huì)有什么像Java 中synchronize的方法了. 那我們?cè)撊绾巫瞿?
1. 盡量不訪問(wèn)同一個(gè)資源. 以避免沖突. 但是可以同時(shí)像數(shù)據(jù)庫(kù)操作. 因?yàn)閿?shù)據(jù)庫(kù)是支持并發(fā)操作的. 所以在多線程的php中
不要向同一個(gè)文件中寫入數(shù)據(jù). 如果必須要寫的話, 用別的方法進(jìn)行同步.. 如調(diào)用 flock對(duì)文件進(jìn)行加鎖等. 或建立臨時(shí)文件并在另外的線程中等待這個(gè)文件的消失 while(file_exits('xxx')); 這樣就等于這個(gè)臨時(shí)文件存在時(shí), 表示其實(shí)線程正在操作,如果沒(méi)有了這個(gè)文件, 說(shuō)明其它線程已經(jīng)釋放了這個(gè).
2. 盡量不要從runThread在執(zhí)行fputs后取這個(gè)socket中讀取數(shù)據(jù). 因?yàn)橐獙?shí)現(xiàn)多線程, 需要的用非阻塞模式. 即在像fgets這樣的函數(shù)時(shí)立即返回.. 所以讀寫數(shù)據(jù)就會(huì)出問(wèn)題. 如果使用阻塞模式的話, 程序就不算是多線程了. 他要等上面的返回才執(zhí)行下面的程序. 所以如果需要交換數(shù)據(jù)最后利用外面文件或數(shù)據(jù)中完成. 實(shí)在想要的話就用socket_set_nonblock($fp) 來(lái)實(shí)現(xiàn).
說(shuō)了這么多, 倒底這個(gè)有沒(méi)有實(shí)際的意義呢? 在什么時(shí)候需要這種用這種方法呢 ?
答案是肯定的. 大家知道. 在一個(gè)不斷讀取網(wǎng)絡(luò)資源的應(yīng)用中, 網(wǎng)絡(luò)的速度是瓶頸. 如果采多這種形式就可以同時(shí)以多個(gè)線程對(duì)不同的頁(yè)面進(jìn)行讀取.
本人做的一個(gè)能從8848、soaso這些商城網(wǎng)站搜索信息的程序。還有一個(gè)從阿里巴巴網(wǎng)站上讀取商業(yè)信息和公司目錄的程序也用到了此技術(shù)。 因?yàn)檫@兩個(gè)程序都是要不斷的鏈接它們的服務(wù)器讀取信息并保存到數(shù)據(jù)庫(kù)。 利用此技術(shù)正好消除了在等待響應(yīng)時(shí)的瓶頸。
多進(jìn)程：使用php的Process Control Functions(PCNTL/線程控制函數(shù))
只能用在Unix Like OS，Windows不可用。
編譯php的時(shí)候，需要加上--enable-pcntl，且推薦僅僅在CLI模式運(yùn)行，不要在WEB服務(wù)器環(huán)境運(yùn)行。
以下為簡(jiǎn)短的測(cè)試代碼：
復(fù)制代碼 代碼如下:
declare(ticks=1);
$bWaitFlag = FALSE; /// 是否等待進(jìn)程結(jié)束
$intNum = 10; /// 進(jìn)程總數(shù)
$pids = array(); /// 進(jìn)程PID數(shù)組
echo ("Start/n");
for($i = 0; $i < $intNum; $i++) {
$pids[$i] = pcntl_fork();/// 產(chǎn)生子進(jìn)程，而且從當(dāng)前行之下開(kāi)試運(yùn)行代碼，而且不繼承父進(jìn)程的數(shù)據(jù)信息
if(!$pids[$i]) {
// 子進(jìn)程進(jìn)程代碼段_Start
$str="";
sleep(5+$i);
for ($j=0;$j<$i;$j++) {$str.="*";}
echo "$i -> " . time() . " $str /n";
exit();
// 子進(jìn)程進(jìn)程代碼段_End
}
}
if ($bWaitFlag)
{
for($i = 0; $i < $intNum; $i++) {
pcntl_waitpid($pids[$i], $status, WUNTRACED);
echo "wait $i -> " . time() . "/n";
}
}
echo ("End/n");
運(yùn)行結(jié)果如下：
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，則結(jié)果如下：
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]$
從 多進(jìn)程的例子可以看出，使用pcntl_fork()之后，將生成一個(gè)子進(jìn)程，而且子進(jìn)程運(yùn)行的代碼，從pcntl_fork()之后的代碼開(kāi)始，而子進(jìn) 程不繼承父進(jìn)程的數(shù)據(jù)信息(實(shí)際上是把父進(jìn)程的數(shù)據(jù)做了一個(gè)全新的拷貝)，因而使用if(!$pids[$i]) 來(lái)控制子進(jìn)程實(shí)際運(yùn)行的代碼段。
更詳細(xì)的研究出于時(shí)間關(guān)系，暫時(shí)沒(méi)有進(jìn)行，你可以參考我給出的手冊(cè)的鏈接。
[文章二] 嘗試php命令行腳本多進(jìn)程并發(fā)執(zhí)行
除了fork, cli下的并發(fā)方式還有一種,看我的例子:
php不支持多線程，但是我們可以把問(wèn)題轉(zhuǎn)換成“多進(jìn)程”來(lái)解決。由于php中的pcntl_fork只有unix平臺(tái)才可以使用，所以本文嘗試使用popen來(lái)替代。
下面是一個(gè)例子：
被并行調(diào)用的子程序代碼：
復(fù)制代碼 代碼如下:
<?php
if($argc==1){
echo("argv/n");
}
$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);
}
}
?>
主調(diào)用者程序，由他調(diào)用子進(jìn)程，同時(shí)并發(fā)的收集子程序的輸出
復(fù)制代碼 代碼如下:
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);
下面是我機(jī)器上的輸出：
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
**總結(jié)：**
**主程序循環(huán)等待子進(jìn)程， 通過(guò)fgets或fread 把子進(jìn)程的輸出獲取出來(lái) ， 從時(shí)間戳上看，的確實(shí)現(xiàn)了并發(fā)執(zhí)行。**
-----------------------------------------------
以后的改進(jìn)：
* popen打開(kāi)的句柄是單向的，如果需要向子進(jìn)程交互，可以使用proc_open
* 使用數(shù)組和子函數(shù)代替while(!feof($handle1)|| !feof($handle2) || !feof($handle3) )這種齷齪的寫法
* 用fread一次把子進(jìn)程已經(jīng)產(chǎn)生的輸出取完，而不是每次一行。
一個(gè)并發(fā)執(zhí)行shell任務(wù)的調(diào)度者,本程序讀取一個(gè)任務(wù)文件,把里面的每行命令并發(fā)執(zhí)行, 可以設(shè)置同時(shí)存在的子進(jìn)程數(shù)目:
復(fù)制代碼 代碼如下:
/*
主任務(wù)管理器
并發(fā)的執(zhí)行子任務(wù)列表
*/
include("../common/conf.php");
include("../common/function.php");
//開(kāi)啟的進(jìn)程數(shù)
$exec_number = 40 ;
/***** main ********/
if($argc==1){
echo("argv/n");
}
$taskfile = $argv[1];
//tasklist
$tasklist = file($taskfile);
$tasklist_len = count($tasklist);
$tasklist_pos = 0;
$handle_list = array();
while(1)
{
//子進(jìn)程列表有空閑,則填充補(bǔ)齊子進(jìn)程列表
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;
}
//如果子進(jìn)程列表空，退出
if(0 == count($handle_list))
{
break;
}
//檢查子進(jìn)程列表的輸出，把停掉的子進(jìn)程關(guān)閉并記錄下來(lái)
$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);
}
}
//踢出停掉的子進(jìn)程
foreach($end_handle_keys as $key)
{
unset($handle_list[$key]);
//var_dump($handle_list);
//exit;
}
}
tolog("/n/n*******************end**********************/n/n", "" , true);
附加一段Socket多進(jìn)程接收的代碼：
復(fù)制代碼 代碼如下:
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);
php技術(shù)：深入解析PHP中的(偽)多線程與多進(jìn)程，轉(zhuǎn)載需保留來(lái)源！
鄭重聲明：本文版權(quán)歸原作者所有，轉(zhuǎn)載文章僅為傳播更多信息之目的，如作者信息標(biāo)記有誤，請(qǐng)第一時(shí)間聯(lián)系我們修改或刪除，多謝。
