• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    公众号
登陆 注册

php模拟并发

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

 

原文： http://blog.csdn.net/zhang_xinglong/article/details/16339867

-------------------------------------------------------------------------------------------------------------------------------

并发请求理论描述：假设有一个client，程序逻辑是要请求三个不同的server，处理各自的响应。传统模型当然是顺序执行，先发送第一个请求，等待收到响应数据后再发送第二个请求，以此类推。就像是单核CPU，一次只能处理一件事，其他事情被暂时阻塞。而并发模式可以让三个server同时处理各自请求，这就可以使大量时间复用。

画个图更好说明问题：

前者为阻塞模式，忽略请求响应等时间，总耗时为700ms；而后者非阻塞模式，由于三个请求可以同时得到处理，总耗时只有300ms。所谓阻塞方式block，顾名思义，就是进程或是线程执行到这些函数时必须等待某个事件的发生，如果事件没有发生，进程或线程就被阻塞，函数不能立即返回。所谓非阻塞方式non-block，就是进程或线程执行此函数时不必非要等待事件的发生，一旦执行肯定返回，以返回值的不同来反映函数的执行情况，如果事件发生则与阻塞方式相同，若事件没有发生则返回一个代码来告知事件未发生，而进程或线程继续执行，所以效率较高。

PHP本身是不支持多线程的，但是它可以利用Linux和apache的多线程能力。php模拟的多线程其实只是多进程，并不是真正的多线程。以下是几种php模拟多线程的方法：
1.php+shell (利用linux os)

php代码(test.php)：

 

[php] view plain copy
 
  1. <?php  
  2. for($i = 0; $i < 10; $i++)  
  3. {  
  4.     echo $i;  
  5.     sleep(5); //这里为了方便看效果sleep一下让脚本执行时间更长  
  6. }  
  7. ?>  

shell代码(test.sh)：

 

 

[plain] view plain copy
 
  1. #!/bin/bash  
  2. for i in 1 2 3 4 5  
  3. do  
  4.     /usr/bin/php -r -q test.php &  
  5. done  

注意：
在请求php代码的那行末尾有一个&符号，这个是关键，不加的话是不能进行多线程的，&表示将服务推送到后台执行，因此在shell的每次的循环中不必等php的代码全部执行完在请求下一个文件，而是同时进行的，这样就实现了多线程，下面运行下shell看下效果，这里你将看到5个test.php进程，再利用linux的定时器，定时请求这个shell，在处理一些需要多线程的任务，例如，批量下载时，非常好用！
参考：http://blog.csdn.net/tianmohust/article/details/8208627

 

2.php+pcntl(利用linux os)
只能用在Unix Like OS，Windows不可用。且推荐仅仅在CLI模式运行，不要在WEB服务器环境运行。

 

[php] view plain copy
 
  1. <?php  
  2. declare(ticks=1);  
  3. //是否等待进程结束  
  4. $bWaitFlag = FALSE;  
  5. //进程总数  
  6. $intNum = 10;  
  7. //进程PID数组  
  8. $pids = array();  
  9. echo ("Start\n");  
  10. for($i = 0; $i < $intNum; $i++)  
  11. {  
  12.     //产生子进程，而且从当前行之下开试运行代码，而且不继承父进程的数据信息  
  13.     $pids[$i] = pcntl_fork();  
  14.     if( ! $pids[$i])  
  15.     {  
  16.         //子进程进程代码段_Start  
  17.         $str = "";  
  18.         sleep(5+$i);  
  19.         for ($j = 0; $j < $i; $j++)  
  20.         {  
  21.             $str .= "*";  
  22.         }  
  23.         echo "$i -> " . time() . " $str \n";  
  24.         exit();  
  25.         //子进程进程代码段_End  
  26.     }  
  27. }  
  28.   
  29. if ($bWaitFlag)  
  30. {  
  31.     for($i = 0; $i < $intNum; $i++)  
  32.     {  
  33.         pcntl_waitpid($pids[$i], $status, WUNTRACED);  
  34.         echo "wait $i -> " . time() . "\n";  
  35.     }  
  36. }  
  37. echo ("End\n");  
  38. ?>  

运行结果如下：

 

 

[plain] view plain copy
 
  1. [qiao@oicq qiao]$ php test.php          
  2. Start  
  3. End  
  4. [qiao@oicq qiao]$ ps -aux | grep "php"  
  5. qiao      32275   0.0   0.5 49668 6148 pts/1     S     14:03    0:00 /usr/local/php4/b  
  6. qiao      32276   0.0   0.5 49668 6152 pts/1     S     14:03    0:00 /usr/local/php4/b  
  7. qiao      32277   0.0   0.5 49668 6152 pts/1     S     14:03    0:00 /usr/local/php4/b  
  8. qiao      32278   0.0   0.5 49668 6152 pts/1     S     14:03    0:00 /usr/local/php4/b  
  9. qiao      32279   0.0   0.5 49668 6152 pts/1     S     14:03    0:00 /usr/local/php4/b  
  10. qiao      32280   0.0   0.5 49668 6152 pts/1     S     14:03    0:00 /usr/local/php4/b  
  11. qiao      32281   0.0   0.5 49668 6152 pts/1     S     14:03    0:00 /usr/local/php4/b  
  12. qiao      32282   0.0   0.5 49668 6152 pts/1     S     14:03    0:00 /usr/local/php4/b  
  13. qiao      32283   0.0   0.5 49668 6152 pts/1     S     14:03    0:00 /usr/local/php4/b  
  14. qiao      32284   0.0   0.5 49668 6152 pts/1     S     14:03    0:00 /usr/local/php4/b  
  15. qiao      32286   0.0   0.0   1620   600 pts/1     S     14:03    0:00 grep php  
  16. [qiao@oicq qiao]$ 0 -> 1133503401    
  17. 1 -> 1133503402 *  
  18. 2 -> 1133503403 **  
  19. 3 -> 1133503404 ***  
  20. 4 -> 1133503405 ****  
  21. 5 -> 1133503406 *****  
  22. 6 -> 1133503407 ******  
  23. 7 -> 1133503408 *******  
  24. 8 -> 1133503409 ********  
  25. 9 -> 1133503410 *********  
  26.   
  27. [qiao@oicq qiao]$  

如果$bWaitFlag=TURE，则结果如下： 

 

 

[plain] view plain copy
 
  1. [qiao@oicq qiao]$ php test.php          
  2. Start  
  3. 0 -> 1133503602    
  4. wait 0 -> 1133503602  
  5. 1 -> 1133503603 *  
  6. wait 1 -> 1133503603  
  7. 2 -> 1133503604 **  
  8. wait 2 -> 1133503604  
  9. 3 -> 1133503605 ***  
  10. wait 3 -> 1133503605  
  11. 4 -> 1133503606 ****  
  12. wait 4 -> 1133503606  
  13. 5 -> 1133503607 *****  
  14. wait 5 -> 1133503607  
  15. 6 -> 1133503608 ******  
  16. wait 6 -> 1133503608  
  17. 7 -> 1133503609 *******  
  18. wait 7 -> 1133503609  
  19. 8 -> 1133503610 ********  
  20. wait 8 -> 1133503610  
  21. 9 -> 1133503611 *********  
  22. wait 9 -> 1133503611  
  23. End  
  24. [qiao@oicq qiao]$   

从多进程的例子可以看出，使用pcntl_fork()之后，将生成一个子进程，而且子进程运行的代码，从pcntl_fork()之后的代码开始，而子进程不继承父进程的数据信息(实际上是把父进程的数据做了一个全新的拷贝)，因而使用if(!$pids[$i]) 来控制子进程实际运行的代码段。
参考：http://hi.baidu.com/tangyubinsir/item/43c04f85ea7709d4d1f8cd84和http://www.itlearner.com/article/4908
3.php+pthreads
参考：http://blog.csdn.net/leinchu/article/details/11795985
4.php+socket(利用web server)
假设你要建立一个服务来检查正在运行的n台服务器，以确定他们还在正常运转。你可能会写下面这样的代码：

[php] view plain copy
 
  1. <?php  
  2. $hosts = array("www.baidu.com", "www.sohu.com", "www.163.com");  
  3.   
  4. $timeout = 15;  
  5. $status = array();  
  6.   
  7. foreach ($hosts as $host)  
  8. {  
  9.     $errno = 0;  
  10.     $errstr = "";  
  11.     $s = fsockopen($host, 80, $errno, $errstr, $timeout);  
  12.     if ($s)  
  13.     {  
  14.         $status[$host] = "Connected\n";  
  15.         fwrite($s, "HEAD / HTTP/1.0\r\nHost: $host\r\n\r\n"); //第二个参数是HTTP协议中规定的请求头，不明白的请看RFC中的定义  
  16.         do  
  17.         {  
  18.             $data = fread($s, 8192);  
  19.             if (strlen($data) == 0)  
  20.             {  
  21.                 break;  
  22.             }  
  23.             $status[$host] .= $data; //返回连接状态  
  24.         }  
  25.         while (true);  
  26.         fclose($s);  
  27.     }  
  28.     else  
  29.     {  
  30.         $status[$host] = "Connection failed: $errno $errstrn";  
  31.     }  
  32. }  
  33. echo \'<pre>\';  
  34. print_r($status);  
  35. ?>  

它运行的很好，但是在fsockopen()分析完hostname并且建立一个成功的连接（或者延时$timeout秒）之前，扩充这段代码来管理大量服务器将耗费很长时间。
因此我们必须放弃这段代码；我们可以建立异步连接-不需要等待fsockopen返回连接状态。PHP仍然需要解析hostname（所以直接使用ip更加明智），不过将在打开一个连接之后立刻返回，继而我们就可以连接下一台服务器。
有两种方法可以实现；PHP5中可以使用新增的stream_socket_client()函数直接替换掉fsocketopen()。PHP5之前的版本，你需要自己动手，用sockets扩展解决问题。
下面是PHP5中的解决方法：

 

 

[php] view plain copy
 
  1. <?php  
  2. $hosts = array("www.baidu.com", "www.sohu.com", "www.163.com");  
  3.   
  4. $timeout = 15;  
  5. $status = array();  
  6.   
  7.   
  8. $sockets = array();  
  9. /* Initiate connections to all the hosts simultaneously */  
  10. foreach ($hosts as $id => $host)   
  11. {  
  12.     $s = stream_socket_client(  
  13.             "$host:80", $errno, $errstr, $timeout,   
  14.             TREAM_CLIENT_ASYNC_CONNECT|STREAM_CLIENT_CONNECT);  
  15.     /* 这里需要稍微延迟一下，否则下面fwrite中的socket句柄不一定能真正使用 
  16.      * 这里应该是PHP的一处bug，查了一下，官方bug早在08年就有人提交了 
  17.      * 我的5.2.8中尚未解决，不知最新的5.3中是否修正 
  18.      */  
  19.     usleep(10);  
  20.     if ($s)   
  21.     {  
  22.         $sockets[$id] = $s;  
  23.         $status[$hosts[$id]] = "in progress";  
  24.     }  
  25.     else   
  26.     {    
  27.         $status[$hosts[$id]] = "failed, $errno $errstr";  
  28.     }  
  29. }  
  30.       
  31. /* Now, wait for the results to come back in */  
  32. while (count($sockets))   
  33. {  
  34.     $read = $write = $sockets;  
  35. //     $e = null;  
  36.     /* This is the magic function - explained below */  
  37.     $n = stream_select($read, $write, $e, $timeout);  
  38.     if ($n > 0) //据说stream_select返回值不总是可信任的    
  39. //     if (count($read))    
  40.     {  
  41.     /* readable sockets either have data for us, or are failed connection attempts */  
  42.         foreach ($read as $r)   
  43.         {  
  44.             /* stream_select generally shuffles $read, so we need to 
  45.              compute from which socket(s) we\'re reading. */  
  46.             $id = array_search($r, $sockets);  
  47.             $data = fread($r, 8192);  
  48.             /* A socket is readable either because it has 
  49.              data to read, OR because it\'s at EOF. */  
  50.             if (strlen($data) == 0)   
  51.             {  
  52.                 if ($status[$hosts[$id]] == "in progress")   
  53.                 {  
  54.                     $status[$hosts[$id]] = "failed to connect";  
  55.                 }  
  56.                 fclose($r);  
  57.                 unset($sockets[$id]);  
  58.             }  
  59.             else   
  60.             {  
  61.                     $status[$hosts[$id]] = $data;  
  62.             }  
  63.         }  
  64.           
  65.         /* writeable sockets can accept an HTTP request */  
  66.         foreach ($write as $w)   
  67.         {  
  68.                 $id = array_search($w, $sockets);  
  69.                 if(is_resource($w) && feof($w) === FALSE)  
  70.                 {  
  71.                     @fwrite($w, "HEAD / HTTP/1.0\r\nHost: " . $hosts[$id] .  "\r\n\r\n");  
  72. //                     $flag && $status[$hosts[$id]] = "waiting for response";  
  73.                 }  
  74.         }  
  75.     }  
  76.     else   
  77.     {  
  78.             /* timed out waiting; assume that all hosts associated with $sockets are faulty */  
  79.             foreach ($sockets as $id => $s)   
  80.             {  
  81.                 $status[$hosts[$id]] = "timed out " . $status[$hosts[$id]];  
  82.             }  
  83.             break;  
  84.     }  
  85. }  
  86.       
  87. echo \'<pre>\';var_dump($status);  
  88. ?>  

我们用stream_select()等待sockets打开的连接事件。stream_select()调用系统的select()函数来工作：前面三个参数是你要使用的streams的数组；你可以对其读取，写入和获取异常（分别针对三个参数）。stream_select()可以通过设置$timeout（秒）参数来等待事件发生-事件发生时，相应的sockets数据将写入你传入的参数。
下面是PHP4.1.0之后版本的实现，如果你已经在编译PHP时包含了sockets(ext/sockets)支持，你可以使用根上面类似的代 码，只是需要将上面的streams/filesystem函数的功能用ext/sockets函数实现。主要的不同在于我们用下面的函数代替 stream_socket_client()来建立连接：

 

 

[php] view plain copy
 
  1. <?php  
  2. // This value is correct for Linux, other systems have other values   
  3. define(\'EINPROGRESS\', 115);    
  4. function non_blocking_connect($host, $port, &$errno, &$errstr, $timeout) {     
  5.         $ip = gethostbyname($host);     
  6.         $s = socket_create(AF_INET, SOCK_STREAM, 0);     
  7.         if (socket_set_nonblock($s)) {      
  8.            $r = @socket_connect($s, $ip, $port);      
  9.            if ($r || socket_last_error() == EINPROGRESS) {       
  10.                   $errno = EINPROGRESS;       
  11.                   return $s;      
  12.                }     
  13.          }     
  14.         $errno = socket_last_error($s);     
  15.         $errstr = socket_strerror($errno);     
  16.         socket_close($s);     
  17.         return false;    
  18. }    
  19. ?>  


现在用socket_select()替换掉stream_select()，用socket_read()替换掉fread()，用socket_write()替换掉fwrite()，用socket_close()替换掉fclose()就可以执行脚本了！ PHP5的先进之处在于，你可以用stream_select()处理几乎所有的stream。例如你可以通过include STDIN用它接收键盘输入并保存进数组，你还可以接收通过proc_open()打开的管道中的数据。
注：select在socket编程中还是比较重要的，可是对于初学socket的人来说都不太爱用select写程序，他们只是习惯写诸如connect、 accept、recv或recvfrom这样的阻塞程序。可是使用select就可以完成非阻塞方式工作的程序，它能够监视我们需要监视的文件描述符的变化情况——读写或是异常。
参考：http://blog.csdn.net/21aspnet/article/details/7420024
5.php+curl
（1）经典curl并发机制和存在问题
经典的cURL实现机制在网上很容易找到, 比如参考PHP在线手册的如下实现方式:

[php] view plain copy
 
  1. <?php  
  2. function classic_curl($urls, $delay) {  
  3.     $queue = curl_multi_init();  
  4.     $map = array();  
  5.   
  6.     foreach ($urls as $url) {  
  7.         // create cURL resources  
  8.         $ch = curl_init();  
  9.   
  10.         // set URL and other appropriate options  
  11.         curl_setopt($ch, CURLOPT_URL, $url);  
  12.   
  13.         curl_setopt($ch, CURLOPT_TIMEOUT, 1);  
  14.         curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);  
  15.         curl_setopt($ch, CURLOPT_HEADER, 0);  
  16.         curl_setopt($ch, CURLOPT_NOSIGNAL, true);  
  17.   
  18.         // add handle  
  19.         curl_multi_add_handle($queue, $ch);  
  20.         $map[$url] = $ch;  
  21.     }  
  22.   
  23.     $active = null;  
  24.   
  25.     // execute the handles  
  26.     do {  
  27.         $mrc = curl_multi_exec($queue, $active);  
  28.     } while ($mrc == CURLM_CALL_MULTI_PERFORM);  
  29.   
  30.     while ($active > 0 && $mrc == CURLM_OK) {  
  31.         if (curl_multi_select($queue, 0.5) != -1) {  
  32.             do {  
  33.                 $mrc = curl_multi_exec($queue, $active);  
  34.             } while ($mrc == CURLM_CALL_MULTI_PERFORM);  
  35.         }  
  36.     }  
  37.   
  38.     $responses = array();  
  39.     foreach ($map as $url=>$ch) {  
  40.         $responses[$url] = callback(curl_multi_getcontent($ch), $delay);  
  41.         curl_multi_remove_handle($queue, $ch);  
  42.         curl_close($ch);  
  43.     }  
  44.   
  45.     curl_multi_close($queue);  
  46.     return $responses;  
  47. }  
  48. ?>  

首先将所有的URL压入并发队列, 然后执行并发过程, 等待所有请求接收完之后进行数据的解析等后续处理. 在实际的处理过程中, 受网络传输的影响, 部分URL的内容会优先于其他URL返回, 但是经典cURL并发必须等待最慢的那个URL返回之后才开始处理, 等待也就意味着CPU的空闲和浪费. 如果URL队列很短, 这种空闲和浪费还处在可接受的范围, 但如果队列很长, 这种等待和浪费将变得不可接受.
（2）改进的rolling curl并发方式
仔细分析不难发现经典cURL并发还存在优化的空间, 优化的方式时当某个URL请求完毕之后尽可能快的去处理它, 边处理边等待其他的URL返回, 而不是等待那个最慢的接口返回之后才开始处理等工作, 从而避免CPU的空闲和浪费. 闲话不多说, 下面贴上具体的实现:

 

 

[php] view plain copy
 
  1. <?php  
  2. function rolling_curl($urls, $delay) {  
  3.     $queue = curl_multi_init();  
  4.     $map = array();  
  5.   
  6.     foreach ($urls as $url) {  
  7.         $ch = curl_init();  
  8.   
  9.         curl_setopt($ch, CURLOPT_URL, $url);  
  10.         curl_setopt($ch, CURLOPT_TIMEOUT, 1);  
  11.         curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);  
  12.         curl_setopt($ch, CURLOPT_HEADER, 0);  
  13.         curl_setopt($ch, CURLOPT_NOSIGNAL, true);  
  14.   
  15.         curl_multi_add_handle($queue, $ch);  
  16.         $map[(string) $ch] = $url;  
  17.     }  
  18.   
  19.     $responses = array();  
  20.     do {  
  21.         while (($code = curl_multi_exec($queue, $active)) == CURLM_CALL_MULTI_PERFORM) ;  
  22.   
  23.         if ($code != CURLM_OK) { break; }  
  24.   
  25.         // a request was just completed -- find out which one  
  26.         while ($done = curl_multi_info_read($queue)) {  
  27.   
  28.             // get the info and content returned on the request  
  29.             $info = curl_getinfo($done[\'handle\']);  
  30.             $error = curl_error($done[\'handle\']);  
  31.             $results = callback(curl_multi_getcontent($done[\'handle\']), $delay);  
  32.             $responses[$map[(string) $done[\'handle\']]] = compact(\'info\', \'error\', \'results\');  
  33.   
  34.             // remove the curl handle that just completed  
  35.             curl_multi_remove_handle($queue, $done[\'handle\']);  
  36.             curl_close($done[\'handle\']);  
  37.         }  
  38.   
  39.         // Block for data in / output; error handling is done by curl_multi_exec  
  40.         if ($active > 0) {  
  41.             curl_multi_select($queue, 0.5);  
  42.         }  
  43.   
  44.     } while ($active);  
  45.   
  46.     curl_multi_close($queue);  
  47.     return $responses;  
  48. }  
  49. ?>  

（3）两种并发实现的性能对比
性能测试中用到的回调函数为:

 

 

[php] view plain copy
 
  1. function callback($data, $delay) {  
  2.     preg_match_all(\'/<h3>(.+)<\/h3>/iU\', $data, $matches);  
  3.     usleep($delay);  
  4.     return compact(\'data\', \'matches\');  
  5. }  

数据处理回调无延迟时: Rolling Curl略优, 但性能提升效果不明显.数据处理回调延迟5毫秒: Rolling Curl完胜, 性能提升40%左右.通过上面的性能对比, 在处理URL队列并发的应用场景中Rolling cURL应该是更加的选择, 并发量非常大(1000+)时, 可以控制并发队列的最大长度, 比如20, 每当1个URL返回并处理完毕之后立即加入1个尚未请求的URL到队列中, 这样写出来的代码会更加健壮, 不至于并发数太大而卡死或崩溃.

鲜花
握手
雷人
路过
鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
More+
上一篇：
简单 PHP + MySQL 数据库动态网站制作 -- 摘抄发布时间：2022-07-10
下一篇：
一分钟搭建 Web 版的 PHP 代码调试利器发布时间：2022-07-10
热门推荐
More+
热门话题
More+
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服（服务时间 9:00～18:00）

在线QQ客服
地址：深圳市南山区西丽大学城创智工业园
电邮：jeky_zhao#qq.com
移动电话：139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap