- void Worker::listen(void)
复制代码用于实例化Worker后执行监听。 此方法主要用于在Worker进程启动后动态创建新的Worker实例,能够实现同一个进程监听多个端口,支持多种协议。需要注意的是用这种方法只是在当前进程增加监听,并不会动态创建新的进程,也不会触发onWorkerStart方法。 例如一个http Worker启动后实例化一个websocket Worker,那么这个进程即能通过http协议访问,又能通过websocket协议访问。由于websocket Worker和http Worker在同一个进程中,所以它们可以访问共同的内存变量,共享所有socket连接。可以做到接收http请求,然后操作websocket客户端完成向客户端推送数据类似的效果。 注意: 如果PHP版本<=7.0,则不支持在多个子进程中实例化相同端口的Worker。例如A进程创建了监听2016端口的Worker,那么B进程就不能再创建监听2016端口的Worker,否则会报Address already in use错误。例如下面的代码是无法运行的。 - use Workerman\Worker;
# G2 s2 f" ^: H, L - require_once __DIR__ . '/Workerman/Autoloader.php';
1 E$ v& g7 W5 ]' k8 ]& R - 5 f1 g$ w6 ~9 X# L- X6 G* t
- $worker = new Worker();: q3 [) C$ R% S! _; x
- // 4个进程& t3 s( s! B4 X# Q
- $worker->count = 4;
6 n- |: q! M' g4 K7 Z - // 每个进程启动后在当前进程新增一个Worker监听
0 \* Q* j+ q/ E - $worker->onWorkerStart = function($worker)
, U. Y( A- u) G; t5 i - {% G; W3 `3 M( t& V0 S
- /**
- d2 Y% V, z& R; S - * 4个进程启动的时候都创建2016端口的Worker! H) v& Y7 P( V7 g6 i
- * 当执行到worker->listen()时会报Address already in use错误
7 q! ~, N; G/ M/ N1 Y3 u - * 如果worker->count=1则不会报错! h( a% Y6 c$ f# C1 B: `
- */" N% ]4 \ Y# G5 y) W( h9 P7 Y8 k
- $inner_worker = new Worker('http://0.0.0.0:2016');2 S- E4 Z6 `% B! |( R0 w% X' ~
- $inner_worker->onMessage = 'on_message';/ K, r z; }9 a5 |) d- e# O
- // 执行监听。这里会报Address already in use错误; q7 b5 h( \6 t4 ^0 D
- $inner_worker->listen();
$ r* [( [8 E) ~3 p- m/ F* r, \' v6 a - };
4 W& P4 d/ s; n2 ]) J - 4 w# x. { t+ ?% q% G: E) k
- $worker->onMessage = 'on_message';, b* Y5 g) b, ^) X
6 h% H+ o7 R, G& C) Z% y3 r- function on_message($connection, $data)
: F9 N/ Q l" Y& T# P& D( B* A# G0 s - {
6 D- o* }/ D* P- c2 X - $connection->send("hello\n");
9 a+ v1 f! [- u4 A; o - }
6 E' v. k5 _% ~
: ?5 w7 C, l+ Y0 f4 W- // 运行worker8 U8 X% j( v0 u- Y
- Worker::runAll();
( M1 @6 K' z- A( `* u+ S; y$ b - 如果您的PHP版本>=7.0,可以设置Worker->reusePort=true, 这样可以做到多个子进程创建相同端口的Worker。见下面的例子:
/ k! S* ^7 M, b. P7 q9 ?
% @% w ~8 b( `8 K5 \% |, b- use Workerman\Worker;
) z0 }5 _, n9 d9 E - require_once './Workerman/Autoloader.php';. ]/ m a$ V! k" H7 j
- + m, X& g4 w/ B0 O7 y' b. n
- $worker = new Worker('text://0.0.0.0:2015');
, n% v0 G r# m2 M - // 4个进程9 d6 Z4 t9 b; ?+ C
- $worker->count = 4;, _. j, C# p2 f# T0 }$ p0 Y0 ^
- // 每个进程启动后在当前进程新增一个Worker监听
- z+ n5 [* l) U$ Z( G. {( G - $worker->onWorkerStart = function($worker)
+ ] f# B! h. Z3 P - {
4 Z2 o( j* J/ Q3 s - $inner_worker = new Worker('http://0.0.0.0:2016');
" r( K! w& Q7 R8 e - // 设置端口复用,可以创建监听相同端口的Worker(需要PHP>=7.0)
$ L/ B! A) m. U4 X7 a - $inner_worker->reusePort = true;. {7 x3 Z! m0 p5 z. W; m9 B
- $inner_worker->onMessage = 'on_message';
5 P. e& P2 s9 Q0 K" v; f, a; O - // 执行监听。正常监听不会报错! l! |2 F. V/ A! U3 q0 G
- $inner_worker->listen();7 X6 l: ~' ?- p: h- F9 U3 v
- };
( K ~- n3 v. D0 ~
- u; o* W! ]& n8 w" i6 k% P- $worker->onMessage = 'on_message';
+ }: p; n3 p# A- U
8 u1 [4 G& k- s2 k7 z- function on_message($connection, $data)4 f8 k* t- }1 _. }, H
- {2 n: w# e# u) m& F
- $connection->send("hello\n");
5 x' L& o0 }" K - }
5 h" M/ c5 l2 s+ d# R' @8 \
+ x7 F2 r6 @6 q6 `0 ~- // 运行worker
8 m1 P5 ]& [ H: W; g* x% ] - Worker::runAll();
复制代码 示例 php后端及时推送消息给客户端原理: 1、建立一个websocket Worker,用来维持客户端长连接 2、websocket Worker内部建立一个text Worker 3、websocket Worker 与 text Worker是同一个进程,可以方便的共享客户端连接 4、某个独立的php后台系统通过text协议与text Worker通讯 5、text Worker操作websocket连接完成数据推送 代码及步骤 push.php - <?php' [7 E3 W/ V3 X& p
- use Workerman\Worker;
, _) l+ \# ^' K, I2 A7 ~7 S - require_once './Workerman/Autoloader.php';
% d. u% _5 @' y - // 初始化一个worker容器,监听1234端口
7 C- n1 u9 w u) U9 W5 }0 |% m - $worker = new Worker('websocket://0.0.0.0:1234');
Z* G+ `4 H5 y, M; b - 2 a8 I) M+ d, ?3 G" l
- /*
( F% c% `' T5 T' Z8 D" J - * 注意这里进程数必须设置为1,否则会报端口占用错误0 m' _& n7 j1 x1 K
- * (php 7可以设置进程数大于1,前提是$inner_text_worker->reusePort=true)
% K7 @ a. k3 J0 I - */
& V2 O, s1 l5 j1 P+ H - $worker->count = 1;5 A- o8 t$ K: {4 w
- // worker进程启动后创建一个text Worker以便打开一个内部通讯端口# u1 s# {" f; m8 R* n3 {0 @/ t
- $worker->onWorkerStart = function($worker)! h1 N* x2 ?7 b3 L7 D" c3 J7 B
- {
% }9 c* q, s0 W, b - // 开启一个内部端口,方便内部系统推送数据,Text协议格式 文本+换行符
7 r" b- b2 y- X J5 @ - $inner_text_worker = new Worker('text://0.0.0.0:5678');
$ T. r2 m: K: d( o: [* j" E - $inner_text_worker->onMessage = function($connection, $buffer)( ?; _5 ~) J; N# T* e
- {
/ G' l2 @& @& `+ C" m - // $data数组格式,里面有uid,表示向那个uid的页面推送数据
% t, ~, F' e6 ?; S - $data = json_decode($buffer, true);
" i, U, p2 ^& V/ S; H8 |& r8 z8 t - $uid = $data['uid'];
; h% L7 u% Z! D - // 通过workerman,向uid的页面推送数据 U! S) |! L$ b4 x; f+ y) K- {
- $ret = sendMessageByUid($uid, $buffer);
) J1 q- H% a# ]0 T! b - // 返回推送结果; |% [9 f0 U( ^6 u; n
- $connection->send($ret ? 'ok' : 'fail');
1 f5 }! Q2 L A0 O - };
6 y8 \; ^; t3 H9 k" Y+ f* x, r3 e& P - // ## 执行监听 ##
2 x8 ` q- B! G6 c3 U+ i - $inner_text_worker->listen();
+ ]# [. ?# [! w; B8 n: y9 ` - };
2 i4 v7 q; [, M6 w4 g" G - // 新增加一个属性,用来保存uid到connection的映射4 Z: n5 ?- u7 x: [+ `* S, d
- $worker->uidConnections = array();
b6 l r0 h$ H, n: A) g7 }8 G, g7 r - // 当有客户端发来消息时执行的回调函数
: Q% ?6 J& _# z1 H ?0 r6 [! `4 [ - $worker->onMessage = function($connection, $data)
# N2 M6 u& L9 A' V" d - {- ?! i1 f- v1 D7 b% z* B8 o
- global $worker;
' [# ^* e1 J6 k( U7 d0 T: c - // 判断当前客户端是否已经验证,既是否设置了uid
# h, i( R& L6 F. {5 F4 v2 V - if(!isset($connection->uid))
: ], v) Z; n: M9 t# L* } - {
' I! h( a3 c" i9 t, x5 O5 t - // 没验证的话把第一个包当做uid(这里为了方便演示,没做真正的验证)1 t0 U6 A% F2 k
- $connection->uid = $data;
- D- z$ G' T4 [* I# u/ \. u - /* 保存uid到connection的映射,这样可以方便的通过uid查找connection, |0 S7 p, Y; |) ?7 c# s3 p
- * 实现针对特定uid推送数据
8 B7 U# i. ?$ c7 b; C- {6 l {; W) x - */) a* u( d+ ~& ^% Q/ o
- $worker->uidConnections[$connection->uid] = $connection;
, H U5 y6 Y8 H; @$ k' c2 z/ b2 f - return;4 Q2 F* Q1 \7 }
- }- e" ~9 U6 F& _; G
- };# O s Q! Z, \/ K5 N0 m5 a) K5 U; q
- * P' o7 d; N" ~! c3 f. f/ X
- // 当有客户端连接断开时; D0 V: N5 s; B; j- i( u
- $worker->onClose = function($connection)) L' c$ i4 q6 w- k! t/ W% h
- {
0 l9 p0 A( T, r+ w5 l - global $worker;3 o) Q) K% v5 S2 S/ v" w8 Z
- if(isset($connection->uid))0 q2 V5 t; p: H
- {: `7 }0 j( {4 F" m8 J, C
- // 连接断开时删除映射
* J* {( z( q" m7 o! ?; r - unset($worker->uidConnections[$connection->uid]);
8 Y) k: G) M% C0 ?! K4 N5 e - }
4 y" v& y( P1 z7 G$ ?) q5 v - };
0 w1 m- E4 Q3 Z* j- Y - & L' V& i# L; z* c6 \
- // 向所有验证的用户推送数据
* _( q h: A* P _/ y2 [ - function broadcast($message)
0 M/ E7 v% I' P# |/ d) j$ {1 W( k - {1 }2 A5 ^* w2 Z: X, [5 j& ?9 R
- global $worker;8 I% F1 B l" N5 O
- foreach($worker->uidConnections as $connection)
$ B1 }6 g2 b8 P; i) Y! m# p* R - {6 X. e& R( K7 x) O# d: U. C; H
- $connection->send($message);0 Q/ E$ M! p o( P- S* X
- }7 v6 m3 @* V, V/ A8 e
- }- u% [8 D [1 N# i
- ' ~+ j- K( M% K0 h8 ?' J# v4 m7 p
- // 针对uid推送数据
4 \' X# g8 m# X* c- g, N8 ~ - function sendMessageByUid($uid, $message)
% F+ R; t0 K8 y - {- u5 ~/ k% X9 q/ i
- global $worker;* |6 G% x4 ~2 h( H( c$ Y( x
- if(isset($worker->uidConnections[$uid]))+ l: o! N/ f3 N6 Y1 y
- {
, B, n4 k. _3 n8 { - $connection = $worker->uidConnections[$uid];4 r5 a' u. E" L6 J8 \- }+ A" c
- $connection->send($message);2 R2 A3 X% e" V7 ?) l
- return true;+ s2 e% @" }$ r9 G i
- }+ e# s% ~ L, b+ n3 W( Q
- return false;
. L# ]( I9 j+ R6 T& v$ j - }) `/ ]3 ~5 V# l: m& t4 q0 U
- 3 J. N% k1 i* |6 r9 ]% s. T- O# ~9 K
- // 运行所有的worker9 A6 P4 y( u+ D/ ?8 Z: H' k
- Worker::runAll();
复制代码启动后端服务 php push.php start -d 前端接收推送的js代码 - var ws = new WebSocket('ws://127.0.0.1:1234');
9 P1 v0 V, [4 b. O( T X, u6 d - ws.onopen = function(){
; f5 s/ e- z0 f! q) j - var uid = 'uid1';9 I+ y8 M6 y3 _+ l% [* N% f9 k
- ws.send(uid);9 M* s8 T d: @$ _! g
- };/ I( c. ^2 R/ m3 r* e. m1 P
- ws.onmessage = function(e){
/ c! ~( N* N$ A+ M. T* A$ V% ]1 l - alert(e.data);
# Z- L' _8 H+ h - };
复制代码后端推送消息的代码 - // 建立socket连接到内部推送端口; { {1 C0 a3 z* u+ m3 s3 {
- $client = stream_socket_client('tcp://127.0.0.1:5678', $errno, $errmsg, 1);4 `) _ K5 \# l- N
- // 推送的数据,包含uid字段,表示是给这个uid推送
* P& p. x$ T& b) d7 p, ~ - $data = array('uid'=>'uid1', 'percent'=>'88%');" H4 F" w, U# u
- // 发送数据,注意5678端口是Text协议的端口,Text协议需要在数据末尾加上换行符
; l8 f- ]( z' k/ I" } - fwrite($client, json_encode($data)."\n");6 K; s/ o* z( v j) Q" t, J7 `
- // 读取推送结果
! v& l7 w- D. o8 D2 `1 c' }# I3 u - echo fread($client, 8192);
复制代码 ! H( K6 G0 Q7 c% e U2 M
X/ E' P6 J' w: n |