php中socket服务的模型下的编程方式(同步和异步)

前面我们花了一段时间来搭建高性能的socket服务，可以同时处理大量的连接，但这是在没有具体业务的情况下。

如果我们启用了一个单进程的server,但里面的一个业务耗时1秒，那么在这1秒内是阻塞的，后续的请求会等待,如果并发三个请求，那么三个请求的执行时间会分别昌1秒,2秒,3秒.提高并发的方法有以下几种:

1:多启动进程，提高并发数

2:优化业务，减少耗时间相当于减少阻塞时间，提高并发数

3:异步编程，避免阻塞，提高并发数

这里我们重点介绍第三种方法,以访问第三方http为例。

代码如下：

1. //同步读取 
2. function get_data_blocking(){ 
3.     $socket = stream_socket_client("tcp://test.raventech.cn:80", $errno, $errstr, 6); 
4.     fwrite($socket, "GET /sleep1.php HTTP/1.0/r/nHost: test.raventech.cn/r/nAccept: */*/r/n/r/n"); 
5.     $str = ""; 
6.     while (!feof($socket)) { 
7.         $str .= fgets($socket, 1024); 
8.     } 
9.     fclose($socket); 
10.     return $str; 
11. } 
12.  
13. //异步读取 
14. function get_data_unblocking(){ 
15.     $socket = stream_socket_client("tcp://test.raventech.cn:80", $errno, $errstr, 6); 
16.     stream_set_blocking($socket, 0); 
17.     fwrite($socket, "GET /sleep1.php HTTP/1.0/r/nHost: test.raventech.cn/r/nAccept: */*/r/n/r/n"); 
18.     $write  = NULL; 
19.     $except = NULL; 
20.     while( $socket ){ 
21.         $read   = array($socket); 
22.         $num_changed_streams = stream_select($read, $write, $except, 0); 
23.         if ( $num_changed_streams > 0 ) { 
24.             foreach($read as $r){ 
25.                 $str = fread($r,2048); 
26.                 fclose($socket); 
27.                 $socket = false; 
28.                 return $str; 
29.             } 
30.         } 
31.         usleep(100); 
32.     } 
33. } 
34.  
35. //真正的异步读取--利用server的IO复用事件来提高并发 
36. class Get_data_event{ 
37.  
38.     public $onMessage = null; 
39.     private $str=''; 
40.  
41.     function __construct(&$server){ 
42.         $socket = stream_socket_client("tcp://test.xtgxiso.cn:80", $errno, $errstr, 6); 
43.         stream_set_blocking($socket, 0); 
44.         fwrite($socket, "GET /sleep1.php HTTP/1.0/r/nHost: test.xtgxiso.cn/r/nAccept: */*/r/n/r/n"); 
45.         $server->add_socket($socket, array($this, 'read')); 
46.     } 
47.  
48.     public function read($socket){ 
49.         while (1) { 
50.             $buffer = fread($socket, 1024); 
51.             if ($buffer === '' || $buffer === false) { 
52.                 break; 
53.             } 
54.             $this->str .= $buffer; 
55.         } 
56.         if( $this->onMessage && $this->str ) { 
57.             call_user_func($this->onMessage, $this->str); 
58.         } 
59.         $this->str = ''; 
60.         return false; 
61.     } 
62.  
63. } 
64.  
65. /** 
66.  * 单进程IO复用select 
67.  */ 
68. class Xtgxiso_server 
69. { 
70.     public $socket = false; 
71.     public $master = array(); 
72.     public $onConnect = null; 
73.     public $onMessage = null; 
74.     public $other_socket_callback = array(); 
75.  
76.     function __construct($host="0.0.0.0",$port=1215) 
77.     { 
78.         $this->socket = stream_socket_server("tcp://".$host.":".$port,$errno, $errstr); 
79.         if (!$this->socket) die($errstr."--".$errno); 
80.         stream_set_blocking($this->socket,0); 
81.         $id = (int)$this->socket; 
82.         $this->master[$id] = $this->socket; 
83.     } 
84.  
85.     public function add_socket($socket,$callback){ 
86.         $id = (int)$socket; 
87.         $this->master[$id] = $socket; 
88.         $this->other_socket_callback[$id] = $callback; 
89.     } 
90.  
91.     public function run(){ 
92.         $read = $this->master; 
93.         $receive = array(); 
94.         echo  "start run.../n"; 
95.         while ( 1 ) { 
96.             $read = $this->master; 
97.             //echo  "waiting.../n"; 
98.             $mod_fd = @stream_select($read, $_w = NULL, $_e = NULL, 60); 
99.             if ($mod_fd === FALSE) { 
100.                 break; 
101.             } 
102.             foreach ( $read as $k => $v ) { 
103.                 $id = (int)$v; 
104.                 if ( $v === $this->socket ) { 
105.                     //echo "new conn/n"; 
106.                     $conn = stream_socket_accept($this->socket); 
107.                     if ($this->onConnect) { 
108.                         call_user_func($this->onConnect, $conn); 
109.                     } 
110.                     $id = (int)$conn; 
111.                     $this->master[$id] = $conn; 
112.                 } else if ( @$this->other_socket_callback[$id] ){ 
113.                     call_user_func_array($this->other_socket_callback[$id], array($v)); 
114.                 } else { 
115.                     //echo "read data/n"; 
116.                     if ( !isset($receive[$k]) ){ 
117.                         $receive[$k]=""; 
118.                     } 
119.                     $buffer = fread($v, 1024); 
120.                     //echo $buffer."/n"; 
121.                     if ( strlen($buffer) === 0 ) { 
122.                         if ( $this->onClose ){ 
123.                             call_user_func($this->onClose,$v); 
124.                         } 
125.                         fclose($v); 
126.                         $id = (int)$v; 
127.                         unset($this->master[$id]); 
128.                     } else if ( $buffer === FALSE ) { 
129.                         if ( $this->onClose ){ 
130.                             call_user_func($this->onClose, $this->master[$key_to_del]); 
131.                         } 
132.                         fclose($v); 
133.                         $id = (int)$v; 
134.                         unset($this->master[$id]); 
135.                     } else { 
136.                         $pos = strpos($buffer, "/r/n/r/n"); 
137.                         if ( $pos === false) { 
138.                             $receive[$k] .= $buffer; 
139.                             //echo "received:".$buffer.";not all package,continue recdiveing/n"; 
140.                         }else{ 
141.                             $receive[$k] .= trim(substr ($buffer,0,$pos+4)); 
142.                             $buffer = substr($buffer,$pos+4); 
143.                             if($this->onMessage) { 
144.                                 call_user_func($this->onMessage,$v,$receive[$k]); 
145.                             } 
146.                             $receive[$k]=''; 
147.                         } 
148.                     } 
149.                 } 
150.             } 
151.             usleep(10000); 
152.         } 
153.     } 
154. } 
155.  
156.  
157. $server =  new Xtgxiso_server(); 
158.  
159. $server->onConnect = function($conn){ 
160.     echo "onConnect -- accepted " . stream_socket_get_name($conn,true) . "/n"; 
161. }; 
162.  
163. $server->onMessage = function($conn,$msg) use ( $server ) { 
164.     /* 
165.     $respone ="";//响应内容 
166.     $respone = "HTTP/1.1 200 OK/r/n"; 
167.     $respone .= "Server: openresty/r/n"; 
168.     $respone .= "Content-Type: text/html; charset=utf-8/r/n"; 
169.     $body = time().rand(111111,999999); 
170.     $len = strlen($body); 
171.     $respone .= "Content-Length:$len/r/n"; 
172.     $respone .= "Connection: close/r/n"; 
173.     $respone .= "/r/n$body/r/n/r/n"; 
174.     echo "onMessage --" . $msg . "/n"; 
175.     */ 
176.  
177.     //同步读取 
178.     //$respone = get_data_blocking(); 
179.     //fwrite($conn,$respone); 
180.  
181.     //异步读取 
182.     //$respone = get_data_unblocking(); 
183.     //fwrite($conn,$respone); 
184.  
185.     //真正异步 
186.     $data = new Get_data_event($server); 
187.     $data->onMessage = function($str) use($conn){ 
188.         fwrite($conn,$str); 
189.     }; 
190.  
191. }; 
192.  
193. $server->onClose = function($conn){ 
194.     echo "onClose --" . "/n"; 
195. }; 
196.  
197. $server->run(); 

第三方服务sleep1.php的代码比较简单:

1. sleep(1);//模拟耗时 
2. echo "OK"; 

通过以上代码示例，我们分别注释运行 同步读取，异步读取,真正异步，来观察server的并发.测试方法可以写个test.html来模拟三个并发.

1. <script src="http://127.0.0.1:1215/?id=1"></script> 
2. <script src="http://127.0.0.1:1215/?id=2"></script> 
3. <script src="http://127.0.0.1:1215/?id=3"></script> 

通过测试发现，真正异步的是并发的，每个请求耗时1秒，这样我们总算明白什么是真正的非阻塞异步编程了，关键就在共用IO复用.