listen

void Worker::listen(void)

用于實(shí)例化Worker后執(zhí)行監(jiān)聽(tīng)。

此方法主要用于在Worker進(jìn)程啟動(dòng)后動(dòng)態(tài)創(chuàng)建新的Worker實(shí)例，能夠?qū)崿F(xiàn)同一個(gè)進(jìn)程監(jiān)聽(tīng)多個(gè)端口，支持多種協(xié)議。需要注意的是用這種方法只是在當(dāng)前進(jìn)程增加監(jiān)聽(tīng)，并不會(huì)動(dòng)態(tài)創(chuàng)建新的進(jìn)程，也不會(huì)觸發(fā)onWorkerStart方法。

例如一個(gè)http Worker啟動(dòng)后實(shí)例化一個(gè)websocket Worker，那么這個(gè)進(jìn)程即能通過(guò)http協(xié)議訪(fǎng)問(wèn)，又能通過(guò)websocket協(xié)議訪(fǎng)問(wèn)。由于websocket Worker和http Worker在同一個(gè)進(jìn)程中，所以它們可以訪(fǎng)問(wèn)共同的內(nèi)存變量，共享所有socket連接?？梢宰龅浇邮説ttp請(qǐng)求，然后操作websocket客戶(hù)端完成向客戶(hù)端推送數(shù)據(jù)類(lèi)似的效果。

注意：

如果PHP版本<=7.0，則不支持在多個(gè)子進(jìn)程中實(shí)例化相同端口的Worker。例如A進(jìn)程創(chuàng)建了監(jiān)聽(tīng)2016端口的Worker，那么B進(jìn)程就不能再創(chuàng)建監(jiān)聽(tīng)2016端口的Worker，否則會(huì)報(bào)Address already in use錯(cuò)誤。例如下面的代碼是無(wú)法運(yùn)行的。

use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker();
// 4個(gè)進(jìn)程
$worker->count = 4;
// 每個(gè)進(jìn)程啟動(dòng)后在當(dāng)前進(jìn)程新增一個(gè)Worker監(jiān)聽(tīng)
$worker->onWorkerStart = function($worker)
{
    /**
     * 4個(gè)進(jìn)程啟動(dòng)的時(shí)候都創(chuàng)建2016端口的Worker
     * 當(dāng)執(zhí)行到worker->listen()時(shí)會(huì)報(bào)Address already in use錯(cuò)誤
     * 如果worker->count=1則不會(huì)報(bào)錯(cuò)
     */
    $inner_worker = new Worker('http://0.0.0.0:2016');
    $inner_worker->onMessage = 'on_message';
    // 執(zhí)行監(jiān)聽(tīng)。這里會(huì)報(bào)Address already in use錯(cuò)誤
    $inner_worker->listen();
};

$worker->onMessage = 'on_message';

function on_message(TcpConnection $connection, $data)
{
    $connection->send("hello\n");
}

// 運(yùn)行worker
Worker::runAll();

如果您的PHP版本>=7.0，可以設(shè)置Worker->reusePort=true， 這樣可以做到多個(gè)子進(jìn)程創(chuàng)建相同端口的Worker。見(jiàn)下面的例子：

use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker('text://0.0.0.0:2015');
// 4個(gè)進(jìn)程
$worker->count = 4;
// 每個(gè)進(jìn)程啟動(dòng)后在當(dāng)前進(jìn)程新增一個(gè)Worker監(jiān)聽(tīng)
$worker->onWorkerStart = function($worker)
{
    $inner_worker = new Worker('http://0.0.0.0:2016');
    // 設(shè)置端口復(fù)用，可以創(chuàng)建監(jiān)聽(tīng)相同端口的Worker（需要PHP>=7.0）
    $inner_worker->reusePort = true;
    $inner_worker->onMessage = 'on_message';
    // 執(zhí)行監(jiān)聽(tīng)。正常監(jiān)聽(tīng)不會(huì)報(bào)錯(cuò)
    $inner_worker->listen();
};

$worker->onMessage = 'on_message';

function on_message(TcpConnection $connection, $data)
{
    $connection->send("hello\n");
}

// 運(yùn)行worker
Worker::runAll();

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

原理：

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

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

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

4、某個(gè)獨(dú)立的php后臺(tái)系統(tǒng)通過(guò)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';

// 初始化一個(gè)worker容器，監(jiān)聽(tīng)1234端口
$worker = new Worker('websocket://0.0.0.0:1234');

/*
 * 注意這里進(jìn)程數(shù)必須設(shè)置為1
 */
$worker->count = 1;
// worker進(jìn)程啟動(dòng)后創(chuàng)建一個(gè)text Worker以便打開(kāi)一個(gè)內(nèi)部通訊端口
$worker->onWorkerStart = function($worker)
{
    // 開(kāi)啟一個(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(TcpConnection $connection, $buffer)
    {
        // $data數(shù)組格式，里面有uid，表示向那個(gè)uid的頁(yè)面推送數(shù)據(jù)
        $data = json_decode($buffer, true);
        $uid = $data['uid'];
        // 通過(guò)workerman，向uid的頁(yè)面推送數(shù)據(jù)
        $ret = sendMessageByUid($uid, $buffer);
        // 返回推送結(jié)果
        $connection->send($ret ? 'ok' : 'fail');
    };
    // ## 執(zhí)行監(jiān)聽(tīng) ##
    $inner_text_worker->listen();
};
// 新增加一個(gè)屬性，用來(lái)保存uid到connection的映射
$worker->uidConnections = array();
// 當(dāng)有客戶(hù)端發(fā)來(lái)消息時(shí)執(zhí)行的回調(diào)函數(shù)
$worker->onMessage = function(TcpConnection $connection, $data)
{
    global $worker;
    // 判斷當(dāng)前客戶(hù)端是否已經(jīng)驗(yàn)證,既是否設(shè)置了uid
    if(!isset($connection->uid))
    {
       // 沒(méi)驗(yàn)證的話(huà)把第一個(gè)包當(dāng)做uid（這里為了方便演示，沒(méi)做真正的驗(yàn)證）
       $connection->uid = $data;
       /* 保存uid到connection的映射，這樣可以方便的通過(guò)uid查找connection，
        * 實(shí)現(xiàn)針對(duì)特定uid推送數(shù)據(jù)
        */
       $worker->uidConnections[$connection->uid] = $connection;
       return;
    }
};

// 當(dāng)有客戶(hù)端連接斷開(kāi)時(shí)
$worker->onClose = function(TcpConnection $connection)
{
    global $worker;
    if(isset($connection->uid))
    {
        // 連接斷開(kāi)時(shí)刪除映射
        unset($worker->uidConnections[$connection->uid]);
    }
};

// 向所有驗(yàn)證的用戶(hù)推送數(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);