php后端及時(shí)推送消息給客戶端

原理：

1、建立一個(gè)websocket Worker，用來維持客戶端長(zhǎng)連接

2、websocket Worker內(nèi)部建立一個(gè)text Worker

3、websocket Worker 與 text Worker是同一個(gè)進(jìn)程，可以方便的共享客戶端連接

4、某個(gè)獨(dú)立的php后臺(tái)系統(tǒng)通過text協(xié)議與text Worker通訊

5、text Worker操作websocket連接完成數(shù)據(jù)推送

代碼及步驟

push.php   <?php use WorkermanWorker; require_once './Workerman/Autoloader.php'; // 初始化一個(gè)worker容器，監(jiān)聽1234端口 $worker = new Worker('websocket://0.0.0.0:1234');//   /*  * 注意這里進(jìn)程數(shù)必須設(shè)置為1，否則會(huì)報(bào)端口占用錯(cuò)誤  * (php 7可以設(shè)置進(jìn)程數(shù)大于1，前提是$inner_text_worker->reusePort=true)  */ $worker->count = 1; // worker進(jìn)程啟動(dòng)后創(chuàng)建一個(gè)text Worker以便打開一個(gè)內(nèi)部通訊端口 $worker->onWorkerStart = function($worker) {     // 開啟一個(gè)內(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($connection, $buffer)     {         // $data數(shù)組格式，里面有uid，表示向那個(gè)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(); }; // 新增加一個(gè)屬性，用來保存uid到connection的映射 $worker->uidConnections = array(); // 當(dāng)有客戶端發(fā)來消息時(shí)執(zhí)行的回調(diào)函數(shù) $worker->onMessage = function($connection, $data) {     global $worker;     // 判斷當(dāng)前客戶端是否已經(jīng)驗(yàn)證,既是否設(shè)置了uid     if(!isset($connection->uid))     {        // 沒驗(yàn)證的話把第一個(gè)包當(dāng)做uid（這里為了方便演示，沒做真正的驗(yàn)證）        $connection->uid = $data;        /* 保存uid到connection的映射，這樣可以方便的通過uid查找connection，         * 實(shí)現(xiàn)針對(duì)特定uid推送數(shù)據(jù)         */        $worker->uidConnections[$connection->uid] = $connection;        return;     } };   // 當(dāng)有客戶端連接斷開時(shí) $worker->onClose = function($connection) {     global $worker;     if(isset($connection->uid))     {         // 連接斷開時(shí)刪除映射         unset($worker->uidConnections[$connection->uid]);     } };   // 向所有驗(yàn)證的用戶推送數(shù)據(jù) function broadcast($message) {    global $worker;    foreach($worker->uidConnections as $connection)    {         $connection->send($message);    } }   // 針對(duì)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; }   // 運(yùn)行所有的worker Worker::runAll(); 啟動(dòng)后端服務(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字段，表示是給這個(gè)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);

后端推送消息的代碼和push.php監(jiān)聽同一個(gè)端口

push.php和前端監(jiān)聽同一個(gè)websocket端口

通過后端推送消息的代碼向push.php推送數(shù)據(jù),

push.php接受到數(shù)據(jù)后通過處理 利用websocket往前端推送數(shù)據(jù)