端口复用技术，这样就可以很好的解决惊群问题和stream_socket_server性能瓶颈的问题.

1. /** 
2.  * 多进程IO复用libevent 
3.  * 同时处理多个连接 
4.  * 端口复用---建议php7 
5.  */ 
6. class Xtgxiso_server 
7. { 
8.     public $socket = false; 
9.     public $master = array(); 
10.     public $onConnect = null; 
11.     public $onMessage = null; 
12.     public $onClose = null; 
13.     public $process_num = 2; 
14.     private $pids = array(); 
15.     public $receive = array(); 
16.     private  $host='127.0.0.1'; 
17.     private $port = 1215; 
18.  
19.     function __construct($host="0.0.0.0",$port=1215){ 
20.         //产生子进程分支 
21.         $pid = pcntl_fork(); 
22.         if ($pid == -1) { 
23.             die("could not fork"); //pcntl_fork返回-1标明创建子进程失败 
24.         } else if ($pid) { 
25.             exit(); //父进程中pcntl_fork返回创建的子进程进程号 
26.         } else { 
27.             // 子进程pcntl_fork返回的时0 
28.         } 
29.         // 从当前终端分离 
30.         if (posix_setsid() == -1) { 
31.             die("could not detach from terminal"); 
32.         } 
33.         umask(0); 
34.         $this->host = $host; 
35.         $this->port = $port; 
36.     } 
37.  
38.     private function start_worker_process(){ 
39.         $pid = pcntl_fork(); 
40.         switch ($pid) { 
41.             case -1: 
42.                 echo "fork error : {$i} \r\n"; 
43.                 exit; 
44.             case 0: 
45.                 $context_option['socket']['so_reuseport'] = 1; 
46.                 $context = stream_context_create($context_option); 
47.                 $this->socket = stream_socket_server("tcp://".$this->host.":".$this->port, $errno, $errstr,STREAM_SERVER_BIND | STREAM_SERVER_LISTEN,$context); 
48.                 if (!$this->socket) die($errstr."--".$errno); 
49.                 stream_set_blocking($this->socket,0); 
50.                 $id = (int)$this->socket; 
51.                 $this->master[$id] = $this->socket; 
52.                 $base = event_base_new(); 
53.                 $event = event_new(); 
54.                 event_set($event, $this->socket, EV_READ | EV_PERSIST, array(__CLASS__, 'ev_accept'), $base); 
55.                 event_base_set($event, $base); 
56.                 event_add($event); 
57.                 echo   posix_getpid()." start run...\n"; 
58.                 event_base_loop($base); 
59.             default: 
60.                 $this->pids[$pid] = $pid; 
61.                 break; 
62.         } 
63.     } 
64.  
65.     public function run(){ 
66.  
67.         for($i = 1; $i <= $this->process_num; $i++){ 
68.             $this->start_worker_process(); 
69.         } 
70.  
71.         while(1){ 
72.             foreach ($this->pids as $i => $pid) { 
73.                 if($pid) { 
74.                     $res = pcntl_waitpid($pid, $status,WNOHANG); 
75.  
76.                     if ( $res == -1 || $res > 0 ){ 
77.                         $this->start_worker_process(); 
78.                         unset($this->pids[$pid]); 
79.                     } 
80.                 } 
81.             } 
82.             sleep(1); 
83.         } 
84.     } 
85.  
86.     public function ev_accept($socket, $flag, $base){ 
87.         $connection = @stream_socket_accept($socket); 
88.         echo posix_getpid()." -- accepted " . stream_socket_get_name($connection,true) . "\n"; 
89.         if ( !$connection ){ 
90.             return; 
91.         } 
92.         stream_set_blocking($connection, 0); 
93.         $id = (int)$connection; 
94.         if($this->onConnect) { 
95.             call_user_func($this->onConnect, $connection); 
96.         } 
97.         $buffer = event_buffer_new($connection, array(__CLASS__, 'ev_read'), array(__CLASS__, 'ev_write'), array(__CLASS__, 'ev_error'), $id); 
98.         event_buffer_base_set($buffer, $base); 
99.         event_buffer_timeout_set($buffer, 30, 30); 
100.         event_buffer_watermark_set($buffer, EV_READ, 0, 0xffffff); 
101.         event_buffer_priority_set($buffer, 10); 
102.         event_buffer_enable($buffer, EV_READ | EV_PERSIST); 
103.         $this->master[$id] = $connection; 
104.         $this->buffer[$id] = $buffer; 
105.         $this->receive[$id] = ''; 
106.     } 
107.  
108.     function ev_read($buffer, $id) 
109.     { 
110.         while( 1 ) { 
111.             $read = event_buffer_read($buffer, 3); 
112.             if($read === '' || $read === false) 
113.             { 
114.                 break; 
115.             } 
116.             $pos = strpos($read, "\n"); 
117.             if($pos === false) 
118.             { 
119.                 $this->receive[$id] .= $read; 
120.                 //echo "received:".$read.";not all package,continue recdiveing\n"; 
121.             }else{ 
122.                 $this->receive[$id] .= trim(substr ($read,0,$pos+1)); 
123.                 $read = substr($read,$pos+1); 
124.                 if($this->onMessage) 
125.                 { 
126.                     call_user_func($this->onMessage,$this->master[$id],$this->receive[$id]); 
127.                 } 
128.                 switch ( $this->receive[$id] ){ 
129.                     case "quit": 
130.                         echo "client close conn\n"; 
131.                         if($this->onClose) { 
132.                             call_user_func($this->onClose, $this->master[$id]); 
133.                         } 
134.                         fclose($this->master[$id]); 
135.                         break; 
136.                     default: 
137.                         //echo "all package:\n"; 
138.                         //echo $this->receive[$id]."\n"; 
139.                         break; 
140.                 } 
141.                 $this->receive[$id]=''; 
142.             } 
143.         } 
144.     } 
145.  
146.     function ev_write($buffer, $id) 
147.     { 
148.         echo "$id -- " ."\n"; 
149.     } 
150.  
151.     function ev_error($buffer, $error, $id) 
152.     { 
153.         echo "ev_error - ".$error."\n"; 
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.     fwrite($conn,"conn success\n"); 
162. }; 
163.  
164. $server->onMessage = function($conn,$msg){ 
165.     echo "onMessage --" . $msg . "\n"; 
166.     fwrite($conn,"received ".$msg."\n"); 
167. }; //phpfensi.com 
168.  
169. $server->onClose = function($conn){ 
170.     echo "onClose --" . stream_socket_get_name($conn,true) . "\n"; 
171. }; 
172.  
173. $server->run(); 

经过多次服务模型的演变，基本我们实现了一个高性能的服务模型！

Tags: socket模型 多进程IO libevent

分享到：