- 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;6 D# r: f2 ~- R6 q' F) h
- require_once __DIR__ . '/Workerman/Autoloader.php';& v5 `% c! F! ] M9 W6 B( k e5 k
9 S' C* _- \5 C7 |0 R, G- @2 U- b- $worker = new Worker();
9 e) P/ a+ T4 D% h' S- W+ \( i - // 4个进程
* S5 E. N/ X; s6 ] C7 n - $worker->count = 4;
' H7 T# q# r" z - // 每个进程启动后在当前进程新增一个Worker监听. ?+ j7 q- E6 l9 `( |
- $worker->onWorkerStart = function($worker)
* ~% @6 s0 w' n( r9 t0 A0 C2 X# A - {
) G( f, l( ], p0 V2 c) {( \1 N, r - /**
6 f3 v+ y2 i u: ^. t - * 4个进程启动的时候都创建2016端口的Worker
) ~7 J) o0 y( r# n, H% u - * 当执行到worker->listen()时会报Address already in use错误# ^3 M& ]; Z3 F% S8 {* Z P* ~
- * 如果worker->count=1则不会报错
+ `. O: ~" d, E% y$ y# f" f9 }* G - */: ^$ X( O! r/ W
- $inner_worker = new Worker('http://0.0.0.0:2016');
: C2 z5 `+ Q7 U' k) v7 l! h - $inner_worker->onMessage = 'on_message';
. q7 B+ \& ?) b9 |* I - // 执行监听。这里会报Address already in use错误# S3 \/ t3 l# Q* B) H E1 M
- $inner_worker->listen();# d1 ` `+ ^! O. c& X8 o
- };
. F- {1 @. v# Q
' h/ ]3 B: o$ J- $worker->onMessage = 'on_message';: _- i* I1 W# i8 M$ D1 K4 N
6 U& w5 c* c* ?- function on_message($connection, $data)
5 v5 ~# j9 C$ i3 k& V: _ - {% |1 {" U; @) T/ R
- $connection->send("hello\n");
" y4 @! P* ]: n1 D5 u/ E - }$ v# [: X1 l$ c& S& ^; _" r
' P- t9 y1 o- R- // 运行worker, |* K+ h5 Z3 g
- Worker::runAll();* L. T0 W' e5 y: u) d& r3 B
- 如果您的PHP版本>=7.0,可以设置Worker->reusePort=true, 这样可以做到多个子进程创建相同端口的Worker。见下面的例子:- u3 |8 c7 c' v3 w0 U4 E# w9 N+ f5 F
# B! S0 i5 ?! J- use Workerman\Worker;
! g/ J1 G6 F; k - require_once './Workerman/Autoloader.php';$ B, n1 ^1 ]' L$ l
- 1 s4 {0 ] m( c) l& ^9 [( T' S% e
- $worker = new Worker('text://0.0.0.0:2015');
' ?. x% U5 k6 f4 B6 D - // 4个进程) n8 z9 ~0 T8 _# N/ v& j) `3 I
- $worker->count = 4;
' O& h: H* o9 b$ G/ ^ - // 每个进程启动后在当前进程新增一个Worker监听
# G% |8 {6 |3 P" ~ - $worker->onWorkerStart = function($worker)
5 d* U7 M8 v s - {
5 G% `1 r/ |0 F: r: M1 @ - $inner_worker = new Worker('http://0.0.0.0:2016');
- H4 w( [1 g& Z- v( Y8 t: [0 @0 s - // 设置端口复用,可以创建监听相同端口的Worker(需要PHP>=7.0)
6 @" b l+ c) X' x: ]' ?- z" Z - $inner_worker->reusePort = true;4 R8 b. Z: @$ x* j" v
- $inner_worker->onMessage = 'on_message';
+ s4 I" H8 c. K) g0 ? - // 执行监听。正常监听不会报错( w' O: V# d. N, E
- $inner_worker->listen();
& |' m3 W" C1 P# v* R% q - }; Q/ [+ x: c0 x* D
/ w0 e. a+ z4 h# n- $worker->onMessage = 'on_message';
! Q. ?. ~. I0 A# b* O* q! ?8 n - . ` P1 M5 g( h- J3 {
- function on_message($connection, $data)
% h& q x& H- A: i; g+ [ - {
8 s8 z3 |2 `/ y, g( b* r - $connection->send("hello\n");; B% D0 K, o% Y8 g7 Q; D: |
- }4 B. H: V% |8 W# i- u" [& U4 j) i
- 8 t! _! |* Z2 }1 h
- // 运行worker3 `# a7 G( S4 ?; j3 ^* o8 G o
- 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
3 L( }# A) A: \; s! P. g6 v2 J - use Workerman\Worker;7 f/ |( i; X& v- Y! U/ [# z' e
- require_once './Workerman/Autoloader.php';- A( `; \2 I7 Y$ {- o/ `) X
- // 初始化一个worker容器,监听1234端口, }, n, p" U: p) h. [' {0 b3 Q, z
- $worker = new Worker('websocket://0.0.0.0:1234');) @% y, H7 ^# d0 R0 |, o2 h
1 {3 L& [( ?% s- m5 u( |- /*
' {- L9 o O, p - * 注意这里进程数必须设置为1,否则会报端口占用错误: \# v: V" e% }
- * (php 7可以设置进程数大于1,前提是$inner_text_worker->reusePort=true)
" H; @( i9 c: X$ ~$ x. o - */+ Q2 Z! |" p; B' j, N
- $worker->count = 1;& [) v- f5 M1 c R" k2 A
- // worker进程启动后创建一个text Worker以便打开一个内部通讯端口
+ f T8 D S: E' [ U- m - $worker->onWorkerStart = function($worker)2 f6 c; C+ N# Y! P
- {9 P7 h7 f6 U5 y
- // 开启一个内部端口,方便内部系统推送数据,Text协议格式 文本+换行符
8 i4 f6 x1 p2 c4 d - $inner_text_worker = new Worker('text://0.0.0.0:5678');+ G5 Q1 j5 i6 V4 r* I
- $inner_text_worker->onMessage = function($connection, $buffer). F0 p: {% M9 s4 N' o+ l! N
- {
" n( e E2 \( `' @8 O- ? - // $data数组格式,里面有uid,表示向那个uid的页面推送数据# K! |: ]8 ~3 C) j5 w
- $data = json_decode($buffer, true);6 E a" V/ O) R: I
- $uid = $data['uid']; C1 U& q+ f" `4 Z
- // 通过workerman,向uid的页面推送数据
; Q4 D; H/ ?# S) W% q3 [ - $ret = sendMessageByUid($uid, $buffer);6 y6 H7 T6 j: F+ b% l
- // 返回推送结果
, C' R5 h' h) s9 c4 D - $connection->send($ret ? 'ok' : 'fail');
# p+ t4 ~3 ]2 ], A - };
4 F' K1 M' s. E( w - // ## 执行监听 ##+ e# Y5 A1 |* F, {/ M
- $inner_text_worker->listen();
4 H4 b" }4 e! ]$ `9 W - };7 P5 n+ n; j6 n- |) ]+ D# o
- // 新增加一个属性,用来保存uid到connection的映射7 F; x6 W: V1 g% C
- $worker->uidConnections = array();- K1 q- ^7 N6 c* H
- // 当有客户端发来消息时执行的回调函数
7 `& s7 g& Z P - $worker->onMessage = function($connection, $data)$ v! `" Q" ^9 k8 h. \, _
- {4 o" x+ U# _6 Y. b/ D2 l' c
- global $worker;8 @9 l0 M% C6 j5 ?
- // 判断当前客户端是否已经验证,既是否设置了uid& G% q! p- {* @0 J
- if(!isset($connection->uid))$ }4 D2 o( J! Y
- {
# Q! W# T8 s- g# C! Q. h - // 没验证的话把第一个包当做uid(这里为了方便演示,没做真正的验证)
6 n, ~8 I- m* x1 }6 U" S - $connection->uid = $data;3 y+ J" X( K2 v
- /* 保存uid到connection的映射,这样可以方便的通过uid查找connection,
( c; x3 f: J/ x- y* Z% F% ] - * 实现针对特定uid推送数据
" ^& k+ K* }! Z8 Q4 P - */
8 |7 O/ x/ y+ E9 H; H1 c3 H - $worker->uidConnections[$connection->uid] = $connection;
% A6 A' P' X! ? | - return;
, t) D6 i8 {8 Y4 I - }' J, y- i( a3 E! A
- };3 p0 {) ]; `! i0 A
) k6 o I% }3 Y* Y5 ]. h& U" g- // 当有客户端连接断开时% n& y1 f, f% m
- $worker->onClose = function($connection)
4 |' e- ]6 O9 x: M, U - {
; y/ U% M1 b: ^# T4 C8 c* e9 g - global $worker;
! w- C/ D8 w1 r - if(isset($connection->uid))
* z0 S4 v# J# e+ R. ] - {1 L4 J0 O" L/ Y+ p7 Q. Y, c3 d
- // 连接断开时删除映射
2 ]5 D g Y4 I' B% g - unset($worker->uidConnections[$connection->uid]);
" T" z) T2 }/ u- L) p A. b9 O - }! J3 J6 r& [, p
- };
# j7 J* c/ p# z) V' ^8 Y - 1 t D; r$ ?+ @+ a- N; t
- // 向所有验证的用户推送数据# d1 a$ k3 S. y2 b1 v
- function broadcast($message)( Q$ x* W' H( P! h- \! G& @: H
- {) v2 N1 q6 Z) p/ |9 R( M
- global $worker;
$ I- L& a7 i! T& [$ w4 I( G- t2 g - foreach($worker->uidConnections as $connection)
/ J+ {& Y1 A/ H( s7 f- i - {$ S" x5 p; w6 P! [* ~8 T! _
- $connection->send($message);" _) X, J1 [* {1 T
- }
( s- {5 D% w. W9 K4 ~! y - }
9 [3 n4 Q I! W( y
; D9 |8 G( Z/ x8 I# `# S( I# [- // 针对uid推送数据
$ P* q& U3 i' {1 A3 V8 n3 p6 ~ - function sendMessageByUid($uid, $message)
/ ^4 t5 E+ |# @. D4 Q+ I - {4 `1 J, f2 z0 S; c( J
- global $worker;
B' G. ~/ z( h5 F/ x- g - if(isset($worker->uidConnections[$uid]))
" K0 T7 u% O3 b% T- R ?" c - {; T M6 H4 ~# K: w1 e0 T
- $connection = $worker->uidConnections[$uid];
" e# S1 x3 z. ~5 a3 B, F - $connection->send($message);
) F7 \9 l! Z0 e9 H" e' G - return true;) \5 y F: _5 Q) f! }3 l+ H0 B( f0 Y
- }
: Z$ r; u e2 g! ~7 @7 R - return false;
$ a3 r5 [7 Q0 M" z. x - }: S* T9 e) N6 W+ u8 H
- % Q1 D' Q$ g& `+ y
- // 运行所有的worker
4 N0 N7 l# V, X3 y( c6 f0 g - Worker::runAll();
复制代码启动后端服务 php push.php start -d 前端接收推送的js代码 - var ws = new WebSocket('ws://127.0.0.1:1234');
* E* Z- |0 L- n N( M - ws.onopen = function(){
' |* W- C' e( ~8 J" g - var uid = 'uid1';
4 @: E' D! N D! o( { - ws.send(uid);# I, j5 V! z8 s! Y
- }; H% R5 }3 o8 g
- ws.onmessage = function(e){7 Y- P! F7 W9 k2 y
- alert(e.data);
" z$ X! C1 G% l% t' s$ x' q# f& L - };
复制代码后端推送消息的代码 - // 建立socket连接到内部推送端口
@' G7 z+ A8 r- t - $client = stream_socket_client('tcp://127.0.0.1:5678', $errno, $errmsg, 1);
3 E3 |2 [6 c3 G3 k4 E* ]' l; L' \ - // 推送的数据,包含uid字段,表示是给这个uid推送$ r8 A, s \4 v7 `/ s2 X& ~+ D5 }
- $data = array('uid'=>'uid1', 'percent'=>'88%');
& ~/ m' Z" C% x8 u. G - // 发送数据,注意5678端口是Text协议的端口,Text协议需要在数据末尾加上换行符
# R$ j1 E) \# _# N4 Q: ~) \ - fwrite($client, json_encode($data)."\n");7 y. T& u' e& c
- // 读取推送结果
, h0 z( ?" Y6 o% b9 r( j% o - echo fread($client, 8192);
复制代码
( m7 N. Z# e' G; H# {7 v& M* L
1 V$ G; s- i, {( A; N9 e |