- 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;9 F9 k3 B5 V% v3 S4 P# i# R; ?
- require_once __DIR__ . '/Workerman/Autoloader.php';5 J+ L, s. V# C% G5 a2 L2 v" G
- - T5 M- u, P9 K- }' w9 Z' v9 M
- $worker = new Worker();8 M; Q N; ^: H& n1 ~
- // 4个进程1 ?# j, F+ n" J8 x+ F7 `: r ?5 q6 h" c
- $worker->count = 4;; {1 H# r8 r3 q; g
- // 每个进程启动后在当前进程新增一个Worker监听% f, Y$ n) A4 T) G# `0 m4 T0 l+ ]
- $worker->onWorkerStart = function($worker)
% C0 O! y: B# Q' m9 Z6 _* ] - {! O5 a" c7 k; N; e& G. z4 c
- /**% A/ }/ w! Y6 J7 g
- * 4个进程启动的时候都创建2016端口的Worker( `7 ]1 u% `1 x# x% [/ U
- * 当执行到worker->listen()时会报Address already in use错误
1 ~+ u; @$ m/ _5 O q7 p - * 如果worker->count=1则不会报错
9 h$ D) j& K- Y: w; m/ N/ j* A - */: o9 Q; ?% O. X7 x
- $inner_worker = new Worker('http://0.0.0.0:2016');
6 x) ?+ p* m; D5 B6 K: e" n - $inner_worker->onMessage = 'on_message';8 u i+ o7 I7 _, R' v2 A" {# ~# Y) s
- // 执行监听。这里会报Address already in use错误
X1 V7 f, w2 b: a/ p& P' U/ t - $inner_worker->listen();
7 c+ b6 X2 H% ? - };
$ S, G- I8 W# l* ]3 g0 F - ( ^; b3 d% y& o' k F
- $worker->onMessage = 'on_message';
P7 {: u( ~6 r( ~$ G/ j/ V0 E
( Y8 Y! k# K0 t/ h8 H0 p) J- function on_message($connection, $data)
% o9 u1 m5 K2 v - {
9 i, ?0 {; F+ l z - $connection->send("hello\n");
5 z7 K0 a: h, N/ j3 q4 f# s - }
7 x s' E; O& c6 @( z
5 j" }' P6 ?$ f x; l0 A8 ~- // 运行worker
$ r3 A! U: L& Z9 t2 t# b9 n - Worker::runAll();' ?! ~( p5 \* \4 v
- 如果您的PHP版本>=7.0,可以设置Worker->reusePort=true, 这样可以做到多个子进程创建相同端口的Worker。见下面的例子:5 ^/ P1 s. d2 K2 m6 q
( X1 I7 n: x) e- use Workerman\Worker;$ R$ B; o o2 [1 _7 J5 X& [- F
- require_once './Workerman/Autoloader.php';2 d* B$ P4 H# Y- q; x4 l
- + ]9 p; j' |; d$ x" O8 M$ R% `
- $worker = new Worker('text://0.0.0.0:2015');
2 `4 j, |& e3 | - // 4个进程+ u1 ^. n) U1 C: J" I' ^
- $worker->count = 4;/ H0 U8 g8 r% h3 S1 W
- // 每个进程启动后在当前进程新增一个Worker监听
}3 m8 Z5 W' y# E - $worker->onWorkerStart = function($worker)
3 N7 X! o4 [& x - {' f% ?( Q; ?+ |/ h7 C
- $inner_worker = new Worker('http://0.0.0.0:2016');: ~7 k; s& L( ]3 Y( N1 F
- // 设置端口复用,可以创建监听相同端口的Worker(需要PHP>=7.0)
* m1 V" }3 R& o/ h - $inner_worker->reusePort = true;1 v" Y0 W6 X) V& r6 h4 z
- $inner_worker->onMessage = 'on_message';
+ K1 K' F. p9 t5 E - // 执行监听。正常监听不会报错8 w5 \( m! O: O
- $inner_worker->listen();
& ^' G; O0 |: I - };/ S; V0 u( U+ |2 p) z& W9 ^
- $ Q: B5 k' ]8 }
- $worker->onMessage = 'on_message';% M$ R4 Q" v5 ?
: F; Z! Y% f! \- function on_message($connection, $data)
- Z+ @4 h& a* i! R: [) ~ Y3 i - { ]/ B% Y8 \ I7 R
- $connection->send("hello\n");' J j* m. D% G/ [" C, q
- }( A5 v/ |' t) D/ ?$ e) G9 t+ ?
- 2 b6 S. a0 D- C; C1 H$ D
- // 运行worker( a0 t3 i! Y+ b( \* q
- 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; ~, p$ ]& u& V
- use Workerman\Worker;5 M' v3 r6 |0 I4 t1 q
- require_once './Workerman/Autoloader.php';* p% l! o; W6 r+ U& x
- // 初始化一个worker容器,监听1234端口
* M% s0 W. O& J# C$ ~3 J - $worker = new Worker('websocket://0.0.0.0:1234');
# U( ^- S5 i* L) b2 m: m9 h, T
# N+ P8 a% y4 J% P! a- /*3 ~* `9 F* U5 L' v. B5 o3 r8 u/ _# I
- * 注意这里进程数必须设置为1,否则会报端口占用错误& d) b8 b9 A7 \- a; D
- * (php 7可以设置进程数大于1,前提是$inner_text_worker->reusePort=true)
6 a+ T- m |6 b - */& p* a: [0 h! y8 b
- $worker->count = 1;6 s5 s: [% a" v% h
- // worker进程启动后创建一个text Worker以便打开一个内部通讯端口
& l6 X4 I/ M, l - $worker->onWorkerStart = function($worker)
6 m' e; E6 N4 n4 v - {
* `/ c# e( ?6 ~- X+ C - // 开启一个内部端口,方便内部系统推送数据,Text协议格式 文本+换行符
4 Y7 o/ a! x8 t7 d! W - $inner_text_worker = new Worker('text://0.0.0.0:5678');
8 q9 E( a) B. R8 h8 l/ I - $inner_text_worker->onMessage = function($connection, $buffer)3 T& I) l4 B- A! m. c2 S4 X
- {/ Q8 N/ B2 e8 }0 y8 i7 Q
- // $data数组格式,里面有uid,表示向那个uid的页面推送数据 a( B# v% W4 R7 U4 P
- $data = json_decode($buffer, true);
' m7 z, ]# C4 P& e# j - $uid = $data['uid'];4 L2 [% ]8 r- i
- // 通过workerman,向uid的页面推送数据; T/ b% a |* b: G3 o5 {/ k9 S
- $ret = sendMessageByUid($uid, $buffer);
R3 [; {9 w! r; @# F; P - // 返回推送结果1 M) T( W4 Q$ G" ] S# c2 J% X
- $connection->send($ret ? 'ok' : 'fail');
) g7 C7 D0 ? p/ `9 p - };/ Y$ W: J2 x7 @
- // ## 执行监听 ##
9 W2 ]( D1 F8 e- [ - $inner_text_worker->listen();2 M% c% Q/ L$ V( l; f3 I8 N
- };& z, @5 n: K- _
- // 新增加一个属性,用来保存uid到connection的映射2 p4 M. f1 h% Y7 y k2 Q4 n
- $worker->uidConnections = array();
% B) |5 Z6 x. t4 h - // 当有客户端发来消息时执行的回调函数
7 ]4 U! X: r/ K- M3 k - $worker->onMessage = function($connection, $data)
6 h8 J" F; A* S6 h1 K5 Y2 I - {2 j( K0 W! B0 u6 q
- global $worker;" D6 [9 L o. b4 _. O
- // 判断当前客户端是否已经验证,既是否设置了uid
6 p2 j# z1 D8 @3 u8 v - if(!isset($connection->uid))/ U+ W G9 {4 v
- {
$ c: t, G/ W( F8 V; K - // 没验证的话把第一个包当做uid(这里为了方便演示,没做真正的验证)8 B' C( [& O" j& G% Q" \8 G
- $connection->uid = $data;
6 w$ |" M7 |" N - /* 保存uid到connection的映射,这样可以方便的通过uid查找connection,
( I G; H! d4 B - * 实现针对特定uid推送数据
, Q5 e- ~- q) C5 N* l - */
8 l+ U+ h" m$ J/ ^) ~5 }$ e - $worker->uidConnections[$connection->uid] = $connection;
4 {$ i; Q' K4 P; N3 v& n - return;0 P/ y- G/ @4 e9 J- d" v
- }2 J6 V! C! Z0 M
- };
( J9 F6 ^$ m4 }2 b - ( S3 l, r" q& T; k, J' ^
- // 当有客户端连接断开时
3 U ^6 T; |9 ?: ]2 @ - $worker->onClose = function($connection)
# \4 ^! Q! f$ _; C* ^9 z - {
* w; B% l5 K( ~1 @ {- F - global $worker;, q; Q% U* D' D, N) L
- if(isset($connection->uid)): d4 m( T$ F% j/ N% M2 j
- {. k* z* _5 r% N5 x4 H
- // 连接断开时删除映射$ p0 Z# W. G# [& w6 N
- unset($worker->uidConnections[$connection->uid]);3 Y% K" J9 {$ d- |& w
- }
; W! Y9 G8 [2 O# {- ^+ |9 {- t - };0 }: G, u; ]3 J; @3 L' S5 b
8 b8 L5 _- B/ i/ G# N- // 向所有验证的用户推送数据
) A& i& i) |; D; l+ c - function broadcast($message)1 Y6 R; ~1 I& D$ n6 L
- {
- P/ Z( R- K9 a; C$ C5 ~ - global $worker;+ G z) j: j7 k5 s& ]
- foreach($worker->uidConnections as $connection)8 ~" Z! i6 j& }: Q
- {
* [( b2 V# @% t! S - $connection->send($message);
. d4 V! D: @1 @" t - }' _4 w" y5 ^4 j% W% E h$ N) J
- }8 S1 e3 m/ N Y4 I" u& p- l! y
- 4 e- U6 G- G- k" n- l+ Z
- // 针对uid推送数据' O# n4 l) M! k# a+ y! [
- function sendMessageByUid($uid, $message)
! j6 d5 Q! r9 m4 I+ M: M; N2 U - {" h1 Z+ x% ~) j9 U& Y$ A
- global $worker;, ~- ]9 S: I) P, D: A+ J
- if(isset($worker->uidConnections[$uid]))
/ @. K" p2 j# M6 Q7 B - {9 q" e+ e) ^- y% q+ J; }
- $connection = $worker->uidConnections[$uid];9 W# Z* V# X7 D0 ~
- $connection->send($message);
% W+ o: ~, s' \+ O. d8 a - return true;: v$ o% ]0 [7 }4 j. X
- }: W; |4 D( U$ T- M" F+ v
- return false;- ] w0 o5 @. |
- }
8 B2 ?3 h H! _- o. c3 V' F6 F
0 y1 Q: C* w# ^0 q4 i- // 运行所有的worker
0 E9 ]7 W& w7 k - Worker::runAll();
复制代码启动后端服务 php push.php start -d 前端接收推送的js代码 - var ws = new WebSocket('ws://127.0.0.1:1234');" f6 D4 L2 ^3 u4 d
- ws.onopen = function(){3 w+ u5 R" L# T+ i
- var uid = 'uid1';) ~; s3 D: u# B! s
- ws.send(uid);6 n; P+ z, n" u0 c( G
- };# H" m, }$ }! o# `
- ws.onmessage = function(e){. R8 W/ g3 [! w8 F' t
- alert(e.data);! X9 V4 }9 v: _6 I, a3 m6 j
- };
复制代码后端推送消息的代码 - // 建立socket连接到内部推送端口
& N3 z1 [+ l! i! ?' ^! k - $client = stream_socket_client('tcp://127.0.0.1:5678', $errno, $errmsg, 1);
9 H- @" @% z0 K, I; J9 | - // 推送的数据,包含uid字段,表示是给这个uid推送/ {9 U! `- z; @3 e7 \
- $data = array('uid'=>'uid1', 'percent'=>'88%');* j4 ]5 e. ]+ J; g
- // 发送数据,注意5678端口是Text协议的端口,Text协议需要在数据末尾加上换行符9 {7 I# a! F/ @: u
- fwrite($client, json_encode($data)."\n");0 u+ y1 E, y" j& o
- // 读取推送结果
# l1 u% {: G4 \: G1 p - echo fread($client, 8192);
复制代码
+ z" s4 ^; a* Y! \! m$ K
6 Z7 Q X# z1 p+ j( \$ V |