php实现websocket实时消息推送

admin

php实现websocket实时消息推送

! k3 @8 R( s2 ?, R- I
; b/ @/ q7 D7 D5 m9 l+ p0 J# JSocketService.php

1. <?php
   : ?, O4 N: v& h4 m
2. /**
   8 D" ?/ h5 ~! P* C7 O5 U, m
3. * Created by xwx
   ( H( d; P# {' X$ b) H. W
4. * Date: 2017/10/18
   
5. * Time: 14:33
   
6. */
   3 t" m8 E. q0 ^6 t. j; I& W
7. 
   : f, g) G) k. k$ ^3 Z* n/ ?
8. class SocketService
   
9. {
   
10.     private $address  = '0.0.0.0';
    3 E) X$ j9 m; d, t# [- l
11.     private $port = 8083;
    & E7 X0 y6 O4 f% T0 `' Y
12.     private $_sockets;
    
13.     public function __construct($address = '', $port='')
    * u6 N- Y! b* \- O9 o4 y7 j
14.     {
    5 T: A& G' e. d- N
15.             if(!empty($address)){
    8 J& l- e* Z0 w  j( L/ t
16.                 $this->address = $address;
    0 W# H! f8 V: n8 V9 ~) [
17.             }
    
18.             if(!empty($port)) {
    
19.                 $this->port = $port;
    2 f. K  C* T; H. B2 [) z, t
20.             }
    2 M3 t$ n4 h; `7 _3 X$ }2 `/ Y; [# x8 @
21.     }
    
22. 
    
23.     public function service(){
    # h/ V% c/ @2 x7 n2 e* @7 q( W
24.         //获取tcp协议号码。
    1 L  F7 K8 u' a
25.         $tcp = getprotobyname("tcp");
    
26.         $sock = socket_create(AF_INET, SOCK_STREAM, $tcp);
    
27.         socket_set_option($sock, SOL_SOCKET, SO_REUSEADDR, 1);
    
28.         if($sock < 0)
    
29.         {
    
30.             throw new Exception("failed to create socket: ".socket_strerror($sock)."\n");
    
31.         }
    3 [( I8 D' c% ~; G$ @/ @
32.         socket_bind($sock, $this->address, $this->port);
    
33.         socket_listen($sock, $this->port);
    
34.         echo "listen on $this->address $this->port ... \n";
    
35.         $this->_sockets = $sock;
    
36.     }
    " Y: S6 ^- q5 F9 z( m* l
37. 
    
38.     public function run(){
    
39.         $this->service();
    
40.         $clients[] = $this->_sockets;
    9 Q: B+ w& |# t
41.         while (true){
    5 p; @8 {. K1 A% G1 E/ g
42.             $changes = $clients;
    2 q! x6 Q. [' |  P
43.             $write = NULL;
    ' w0 y7 r( b9 E& c$ y$ @
44.             $except = NULL;
    
45.             socket_select($changes,  $write,  $except, NULL);
    8 z7 p. V. h8 {/ s4 R7 p9 k
46.             foreach ($changes as $key => $_sock){
    - ^9 m5 Y2 U* r5 K" z+ f
47.                 if($this->_sockets == $_sock){ //判断是不是新接入的socket
    6 p& X' y& T7 l# U
48.                     if(($newClient = socket_accept($_sock))  === false){
    - E" l6 |9 e! S7 v
49.                         die('failed to accept socket: '.socket_strerror($_sock)."\n");
    
50.                     }
      _) Q6 S! q7 C. r. p
51.                     $line = trim(socket_read($newClient, 1024));
    . K1 a+ n7 |3 A! C
52.                     $this->handshaking($newClient, $line);
    . Y( H8 |+ U- t. \! S
53.                     //获取client ip
    ' a: Y  f- s& F0 d- n( r
54.                     socket_getpeername ($newClient, $ip);
    
55.                     $clients[$ip] = $newClient;
    " ]) F! m. q" M, s
56.                     echo  "Client ip:{$ip}   \n";
    3 c' r5 F0 R8 P" r2 I
57.                     echo "Client msg:{$line} \n";
    ! n! e( l5 ?0 ^) p6 h5 b
58.                 } else {
    
59.                     socket_recv($_sock, $buffer,  2048, 0);
    8 j) e; K6 ~  S/ E; G8 r, Y( p* n
60.                     $msg = $this->message($buffer);
    : b& L" L! L7 P* y
61.                     //在这里业务代码
    5 w% s3 Q2 P/ F4 {
62.                     echo "{$key} clinet msg:",$msg,"\n";
    
63.                     fwrite(STDOUT, 'Please input a argument:');
    * P, Z+ T" X: F+ t5 X* R$ ]
64.                     $response = trim(fgets(STDIN));
    
65.                     $this->send($_sock, $response);
    
66.                     echo "{$key} response to Client:".$response,"\n";
    
67.                 }
    
68.             }
    : \) V+ f) Z8 X: c% q; q3 u1 Z
69.         }
    / a  R, @: B. B9 n% ^& |+ z
70.     }
    
71. 
    
72.     /**
    . p: a. ~0 \3 d! A7 J' L
73.      * 握手处理
    
74.      * @param $newClient socket
    2 e$ V8 L; Z8 F! h8 f8 m3 q9 A# F/ n& j
75.      * @return int  接收到的信息
    - [6 H! G4 G" i& Q* A
76.      */
    8 K1 W/ w) q+ ]. Y1 e
77.     public function handshaking($newClient, $line){
    
78. 
    2 e) l0 y/ ?; s- S. M. ?
79.         $headers = array();
    3 f. q% X+ Q! C' l7 v
80.         $lines = preg_split("/\r\n/", $line);
    $ g( Y/ \' M% P! Q0 O
81.         foreach($lines as $line)
    6 ~3 w6 v4 o# g. a5 L* Z* e& a3 i
82.         {
    ' T. |! Y2 \. [/ L3 ^
83.             $line = chop($line);
    
84.             if(preg_match('/\A(\S+): (.*)\z/', $line, $matches))
    ' h1 c& I2 u- D
85.             {
    3 o) p/ a# U' {! Z; |
86.                 $headers[$matches[1]] = $matches[2];
    
87.             }
    
88.         }
    8 n1 D' q* V7 ^! D0 K* o
89.         $secKey = $headers['Sec-WebSocket-Key'];
      B1 j4 Q: d( B. @3 G; m/ e: {' b
90.         $secAccept = base64_encode(pack('H*', sha1($secKey . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11')));
    * c9 w& O9 Q+ k3 ^
91.         $upgrade  = "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" .
    
92.             "Upgrade: websocket\r\n" .
    
93.             "Connection: Upgrade\r\n" .
    
94.             "WebSocket-Origin: $this->address\r\n" .
    
95.             "WebSocket-Location: ws://$this->address:$this->port/websocket/websocket\r\n".
    
96.             "Sec-WebSocket-Accept:$secAccept\r\n\r\n";
    ; v0 k8 u- I  S8 [6 f
97.         return socket_write($newClient, $upgrade, strlen($upgrade));
    
98.     }
    
99. 
    1 T9 M0 p% [3 u' }9 P
100.     /**
     
101.      * 解析接收数据
     
102.      * @param $buffer
     
103.      * @return null|string
     
104.      */
     
105.     public function message($buffer){
     ; ]" J% L" X5 H% M& o% c2 Q
106.         $len = $masks = $data = $decoded = null;
     
107.         $len = ord($buffer[1]) & 127;
     - ?, H. [4 d& m
108.         if ($len === 126)  {
     
109.             $masks = substr($buffer, 4, 4);
     + z0 U/ z6 r  C2 d- K0 x
110.             $data = substr($buffer, 8);
     , U% k+ y$ C/ W0 f
111.         } else if ($len === 127)  {
     
112.             $masks = substr($buffer, 10, 4);
     
113.             $data = substr($buffer, 14);
     + N9 C0 E2 ~8 @: g/ G
114.         } else  {
     
115.             $masks = substr($buffer, 2, 4);
     4 d. J" a, E" q8 W
116.             $data = substr($buffer, 6);
     
117.         }
     6 i+ A* q, Z1 a' Z$ w  j/ g# M
118.         for ($index = 0; $index < strlen($data); $index++) {
     
119.             $decoded .= $data[$index] ^ $masks[$index % 4];
     
120.         }
     
121.         return $decoded;
     
122.     }
     
123. 
     # ^: x2 y4 I! O6 j: H, A" Y" |
124.     /**
     0 F* J5 L+ W3 D. j8 [- A+ u/ k
125.      * 发送数据
     
126.      * @param $newClinet 新接入的socket
     6 r8 A# V$ C& h  y' {! K
127.      * @param $msg   要发送的数据
     
128.      * @return int|string
     
129.      */
       }1 V0 E% H, f2 S5 s# \
130.     public function send($newClinet, $msg){
     / S% V% E" B1 @* N
131.         $msg = $this->frame($msg);
     2 N5 y% t  Z0 S
132.         socket_write($newClinet, $msg, strlen($msg));
     0 i% a; {: o0 Y1 A) z4 ?2 D: P  \
133.     }
     ; l4 Z$ s4 d: E- J9 d) F
134. 
     
135.     public function frame($s) {
     & _/ u; k9 v( x' I$ i
136.         $a = str_split($s, 125);
     % Z* G0 a5 c4 Q8 L8 q: M! T; b+ G
137.         if (count($a) == 1) {
     
138.             return "\x81" . chr(strlen($a[0])) . $a[0];
     
139.         }
     + g! c# V3 `4 E$ `
140.         $ns = "";
     
141.         foreach ($a as $o) {
     
142.             $ns .= "\x81" . chr(strlen($o)) . $o;
     
143.         }
     4 Y5 J3 b$ ]4 x6 t7 R
144.         return $ns;
     + |6 p1 j; h2 e& j4 o
145.     }
     
146. 
     : j$ O' ]% R2 i
147.     /**
     9 j8 Y2 a( q9 o0 ~. G
148.      * 关闭socket
     % X1 W, \. ]* _$ I; D0 q( X
149.      */
     
150.     public function close(){
     , `8 E3 U1 s  G! e3 m
151.         return socket_close($this->_sockets);
     
152.     }
     & n8 \2 ]2 ^. ~( ]2 e" R
153. }
     + O6 A9 o& W  M& ]( n4 p
154. 
     # M0 e3 K$ Q5 u7 Q
155. $sock = new SocketService();
     
156. $sock->run();
     
157. 
     

复制代码

web.html
8 q$ Y3 t& l' t2 X8 W& l# i: i

1. <!doctype html>
   * p: }0 O1 Y7 ]* |2 g9 d; s
2. <html lang="en">
   : {( W: k$ v: A1 n" v8 A" z
3. <head>
   ; G# X* Q& D; N
4.   <meta charset="UTF-8">
   ; e% I% f4 o0 t3 h. E0 _$ u
5.   <meta name="viewport" content="width=device-width,initial-scale=1, maximum-scale=1, user-scalable=no">
   . _+ R' R. u% Y' v% K# n: k6 I
6.   <title>websocket</title>
   
7. </head>
   
8. <body>
   
9. <input id="text" value="">
   - k# ~  E" m3 `& V
10. <input type="submit" value="send" onclick="start()">
    ) g- X" q" o/ J0 p4 V! K$ [1 p0 `
11. <input type="submit" value="close" onclick="close()">
    3 Y8 M" k0 W6 e, f+ e
12. <div id="msg"></div>
    
13. <script>
    
14. /**
    1 ]0 w; _0 L: B: X$ W
15. 0：未连接
    / q& ~' p3 J" E" f
16. 1：连接成功，可通讯
    
17. 2：正在关闭
    $ g" a! }5 ]# x, \2 L/ w/ U
18. 3：连接已关闭或无法打开
    
19. */
    
20. 
    
21.     //创建一个webSocket 实例
    
22.     var webSocket  = new  WebSocket("ws://192.168.31.152:8083");
    2 r/ h8 X/ s  h" U3 _
23. 
    + \, \" v- o6 B4 d. F$ j
24. 
    
25.     webSocket.onerror = function (event){
    $ G0 U5 {5 R' d$ i# |* d& C
26.         onError(event);
    
27.     };
    
28. 
    
29.     // 打开websocket
    
30.     webSocket.onopen = function (event){
    
31.         onOpen(event);
    
32.     };
    ; w2 s. ]* `6 c5 U0 C
33. 
    
34.     //监听消息
    
35.     webSocket.onmessage = function (event){
    2 |8 m( l3 S4 Q, A+ V
36.         onMessage(event);
    : s1 S9 ]' J. U6 b3 v  X
37.     };
    
38. 
    % q. _# d2 P  v" q/ U4 p$ C# P2 a
39. 
    ; ^$ A3 J( ^5 v* Y; ?7 ?
40.     webSocket.onclose = function (event){
    
41.         onClose(event);
    
42.     }
    ( x4 H9 X5 s! o' @. X
43. 
    3 x# E3 B8 }: _% e9 H
44.     //关闭监听websocket
    
45.     function onError(event){
    6 R$ g. T: i' a
46.         document.getElementById("msg").innerHTML = "<p>close</p>";
    6 ~$ K8 k' j- U6 Q# i/ K
47.         console.log("error"+event.data);
    
48.     };
    0 v  ?& e4 k. z5 d) d
49. 
    
50.     function onOpen(event){
    
51.         console.log("open:"+sockState());
    " U) q5 }% n, O
52.         document.getElementById("msg").innerHTML = "<p>Connect to Service</p>";
    
53.     };
    
54.     function onMessage(event){
    * S0 O+ {" h$ I: D
55.         console.log("onMessage");
    
56.         document.getElementById("msg").innerHTML += "<p>response:"+event.data+"</p>"
    / L4 H; G# a& c* }/ L
57.     };
    
58. 
    , k( x3 g) ?, I/ {0 y' U: v* \
59.     function onClose(event){
    
60.         document.getElementById("msg").innerHTML = "<p>close</p>";
    
61.         console.log("close:"+sockState());
    3 |# W3 U, E) |2 E* ?4 }! u& E9 E
62.         webSocket.close();
    2 _; L% s9 u4 S4 D* A
63.     }
    
64. 
    1 x9 c0 x0 I) [6 B9 n
65.     function sockState(){
    
66.         var status = ['未连接','连接成功，可通讯','正在关闭','连接已关闭或无法打开'];
    * E. B- r: B7 m4 [8 E) \; L5 D
67.             return status[webSocket.readyState];
    - [4 _; \) X# [: x7 F5 |
68.     }
    
69. 
    
70. 
    / s: |' H' n% a2 P' C/ ]4 j
71. 
    
72. function start(event){
    / c" d6 P* q) p7 _  Z. z
73.         console.log(webSocket);
    4 e% f- }. a# U& ]& _: Z
74.         var msg = document.getElementById('text').value;
    
75.         document.getElementById('text').value = '';
    
76.         console.log("send:"+sockState());
    7 ~. ]* S/ z* C( J1 H
77.         console.log("msg="+msg);
    
78.         webSocket.send("msg="+msg);
    / G: o7 A, O) E! B( K! H. ?
79.         document.getElementById("msg").innerHTML += "<p>request"+msg+"</p>"
    ; `! `, ]9 M6 I# I9 a
80.     };
    
81. 
    9 n" a# Z1 k7 p2 b
82.     function close(event){
    
83.         webSocket.close();
    % n+ `; ~3 h; G& i
84.     }
    9 ~$ R/ [" {3 b
85. </script>
    
86. </body>
    
87. </html>

复制代码

2 n' t0 l, y/ J4 H. Y! U