listen
void Worker::listen(void)
用于實例化Worker后執(zhí)行監(jiān)聽。
此方法主要用于在Worker進程啟動后動態(tài)創(chuàng)建新的Worker實例,能夠?qū)崿F(xiàn)同一個進程監(jiān)聽多個端口,支持多種協(xié)議。需要注意的是用這種方法只是在當(dāng)前進程增加監(jiān)聽,并不會動態(tài)創(chuàng)建新的進程,也不會觸發(fā)onWorkerStart方法。
例如一個http Worker啟動后實例化一個websocket Worker,那么這個進程即能通過http協(xié)議訪問,又能通過websocket協(xié)議訪問。由于websocket Worker和http Worker在同一個進程中,所以它們可以訪問共同的內(nèi)存變量,共享所有socket連接。可以做到接收http請求,然后操作websocket客戶端完成向客戶端推送數(shù)據(jù)類似的效果。
注意:
如果PHP版本<=7.0,則不支持在多個子進程中實例化相同端口的Worker。例如A進程創(chuàng)建了監(jiān)聽2016端口的Worker,那么B進程就不能再創(chuàng)建監(jiān)聽2016端口的Worker,否則會報Address already in use
錯誤。例如下面的代碼是無法
運行的。
use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';
$worker = new Worker();
// 4個進程
$worker->count = 4;
// 每個進程啟動后在當(dāng)前進程新增一個Worker監(jiān)聽
$worker->onWorkerStart = function($worker)
{
/**
* 4個進程啟動的時候都創(chuàng)建2016端口的Worker
* 當(dāng)執(zhí)行到worker->listen()時會報Address already in use錯誤
* 如果worker->count=1則不會報錯
*/
$inner_worker = new Worker('http://0.0.0.0:2016');
$inner_worker->onMessage = 'on_message';
// 執(zhí)行監(jiān)聽。這里會報Address already in use錯誤
$inner_worker->listen();
};
$worker->onMessage = 'on_message';
function on_message(TcpConnection $connection, $data)
{
$connection->send("hello\n");
}
// 運行worker
Worker::runAll();
如果您的PHP版本>=7.0,可以設(shè)置Worker->reusePort=true, 這樣可以做到多個子進程創(chuàng)建相同端口的Worker。見下面的例子:
use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';
$worker = new Worker('text://0.0.0.0:2015');
// 4個進程
$worker->count = 4;
// 每個進程啟動后在當(dāng)前進程新增一個Worker監(jiān)聽
$worker->onWorkerStart = function($worker)
{
$inner_worker = new Worker('http://0.0.0.0:2016');
// 設(shè)置端口復(fù)用,可以創(chuàng)建監(jiān)聽相同端口的Worker(需要PHP>=7.0)
$inner_worker->reusePort = true;
$inner_worker->onMessage = 'on_message';
// 執(zhí)行監(jiān)聽。正常監(jiān)聽不會報錯
$inner_worker->listen();
};
$worker->onMessage = 'on_message';
function on_message(TcpConnection $connection, $data)
{
$connection->send("hello\n");
}
// 運行worker
Worker::runAll();
示例 php后端及時推送消息給客戶端
原理:
1、建立一個websocket Worker,用來維持客戶端長連接
2、websocket Worker內(nèi)部建立一個text Worker
3、websocket Worker 與 text Worker是同一個進程,可以方便的共享客戶端連接
4、某個獨立的php后臺系統(tǒng)通過text協(xié)議與text Worker通訊
5、text Worker操作websocket連接完成數(shù)據(jù)推送
代碼及步驟
push.php
<?php
use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';
// 初始化一個worker容器,監(jiān)聽1234端口
$worker = new Worker('websocket://0.0.0.0:1234');
/*
* 注意這里進程數(shù)必須設(shè)置為1
*/
$worker->count = 1;
// worker進程啟動后創(chuàng)建一個text Worker以便打開一個內(nèi)部通訊端口
$worker->onWorkerStart = function($worker)
{
// 開啟一個內(nèi)部端口,方便內(nèi)部系統(tǒng)推送數(shù)據(jù),Text協(xié)議格式 文本+換行符
$inner_text_worker = new Worker('text://0.0.0.0:5678');
$inner_text_worker->onMessage = function(TcpConnection $connection, $buffer)
{
// $data數(shù)組格式,里面有uid,表示向那個uid的頁面推送數(shù)據(jù)
$data = json_decode($buffer, true);
$uid = $data['uid'];
// 通過workerman,向uid的頁面推送數(shù)據(jù)
$ret = sendMessageByUid($uid, $buffer);
// 返回推送結(jié)果
$connection->send($ret ? 'ok' : 'fail');
};
// ## 執(zhí)行監(jiān)聽 ##
$inner_text_worker->listen();
};
// 新增加一個屬性,用來保存uid到connection的映射
$worker->uidConnections = array();
// 當(dāng)有客戶端發(fā)來消息時執(zhí)行的回調(diào)函數(shù)
$worker->onMessage = function(TcpConnection $connection, $data)
{
global $worker;
// 判斷當(dāng)前客戶端是否已經(jīng)驗證,既是否設(shè)置了uid
if(!isset($connection->uid))
{
// 沒驗證的話把第一個包當(dāng)做uid(這里為了方便演示,沒做真正的驗證)
$connection->uid = $data;
/* 保存uid到connection的映射,這樣可以方便的通過uid查找connection,
* 實現(xiàn)針對特定uid推送數(shù)據(jù)
*/
$worker->uidConnections[$connection->uid] = $connection;
return;
}
};
// 當(dāng)有客戶端連接斷開時
$worker->onClose = function(TcpConnection $connection)
{
global $worker;
if(isset($connection->uid))
{
// 連接斷開時刪除映射
unset($worker->uidConnections[$connection->uid]);
}
};
// 向所有驗證的用戶推送數(shù)據(jù)
function broadcast($message)
{
global $worker;
foreach($worker->uidConnections as $connection)
{
$connection->send($message);
}
}
// 針對uid推送數(shù)據(jù)
function sendMessageByUid($uid, $message)
{
global $worker;
if(isset($worker->uidConnections[$uid]))
{
$connection = $worker->uidConnections[$uid];
$connection->send($message);
return true;
}
return false;
}
// 運行所有的worker
Worker::runAll();
啟動后端服務(wù)
php push.php start -d
前端接收推送的js代碼
var ws = new WebSocket('ws://127.0.0.1:1234');
ws.onopen = function(){
var uid = 'uid1';
ws.send(uid);
};
ws.onmessage = function(e){
alert(e.data);
};
后端推送消息的代碼
// 建立socket連接到內(nèi)部推送端口
$client = stream_socket_client('tcp://127.0.0.1:5678', $errno, $errmsg, 1);
// 推送的數(shù)據(jù),包含uid字段,表示是給這個uid推送
$data = array('uid'=>'uid1', 'percent'=>'88%');
// 發(fā)送數(shù)據(jù),注意5678端口是Text協(xié)議的端口,Text協(xié)議需要在數(shù)據(jù)末尾加上換行符
fwrite($client, json_encode($data)."\n");
// 讀取推送結(jié)果
echo fread($client, 8192);