您当前的位置： 首页 > 慢生活 > 程序人生 网站首页程序人生

workerman常用组件-channel分佈式通訊組件-channelClient-例子-集群推送

发布时间：2021-12-19 16:18:20编辑：雪饮阅读（）

 例子1

(要求Workerman版本>=3.3.0)
基于Worker的多进程(分布式集群)推送系统，集群群发、集群广播。
首先我們假定有43.170這個内網服務作爲channel服務端并且啓動channel服務：

服務脚本：
<?php
use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';

// 初始化一个Channel服务端
$channel_server = new Channel\Server('0.0.0.0', 2206);

Worker::runAll();
然後在43.42服務器上部署webSocket服務：
該服務脚本用於處理websocket消息經發送到當前脚本對應進程上某個目標websocket鏈接的目標用戶或者做廣播轉發，用來接收一些channel上面的廣播事件並轉發給當前進程的所有websocket客戶端。
環境準備：
[root@localhost ~]# php /root/composer.phar config -g repo.packagist composer https://mirrors.aliyun.com/composer/
Do not run Composer as root/super user! See https://getcomposer.org/root for details
Continue as root/super user [yes]? yes
[root@localhost ~]# php /root/composer.phar require workerman/workerman
Do not run Composer as root/super user! See https://getcomposer.org/root for details
Continue as root/super user [yes]? yes
Using version ^4.0 for workerman/workerman
./composer.json has been updated
Running composer update workerman/workerman
Loading composer repositories with package information
Updating dependencies
Lock file operations: 1 install, 0 updates, 0 removals
 - Locking workerman/workerman (v4.0.25)
Writing lock file
Installing dependencies from lock file (including require-dev)
Package operations: 1 install, 0 updates, 0 removals
 - Downloading workerman/workerman (v4.0.25)
 - Installing workerman/workerman (v4.0.25): Extracting archive
1 package suggestions were added by new dependencies, use `composer suggest` to see details.
Generating autoload files
[root@localhost ~]# php /root/composer.phar require workerman/channel
Do not run Composer as root/super user! See https://getcomposer.org/root for details
Continue as root/super user [yes]? yes
Using version ^1.1 for workerman/channel
./composer.json has been updated
Running composer update workerman/channel
Loading composer repositories with package information
Updating dependencies
Lock file operations: 1 install, 0 updates, 0 removals
 - Locking workerman/channel (v1.1.0)
Writing lock file
Installing dependencies from lock file (including require-dev)
Package operations: 1 install, 0 updates, 0 removals
 - Downloading workerman/channel (v1.1.0)
 - Installing workerman/channel (v1.1.0): Extracting archive
Generating autoload files
服務端脚本：

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

// websocket服务端
$worker = new Worker('websocket://0.0.0.0:4236');
$worker->count=2;
$worker->name = 'pusher';
$worker->onWorkerStart = function($worker)
{
    // Channel客户端连接到Channel服务端
    Channel\Client::connect('192.168.43.170', 2206);
    // 以自己的进程id为事件名称
    $event_name = $worker->id;
   // 订阅worker->id事件并注册事件处理函数
    Channel\Client::on($event_name, function($event_data)use($worker){
        $to_connection_id = $event_data['to_connection_id'];
        $message = $event_data['content'];
        if(!isset($worker->connections[$to_connection_id]))
        {
            echo "connection not exists\n";
            return;
        }
        $to_connection = $worker->connections[$to_connection_id];
        $to_connection->send($message);
    });

    // 订阅广播事件
    $event_name = '广播';
    // 收到广播事件后向当前进程内所有客户端连接发送广播数据
    Channel\Client::on($event_name, function($event_data)use($worker){
        $message = $event_data['content'];
        foreach($worker->connections as $connection)
        {
            $connection->send($message);
       }
    });
};

$worker->onConnect = function(TcpConnection $connection)use($worker)
{
    $msg = "workerID:{$worker->id} connectionID:{$connection->id} connected\n";
    echo $msg;
    $connection->send($msg);
};
Worker::runAll();

服務端啓動：

[root@localhost ~]# php wsServer.php start
Workerman[wsServer.php] start in DEBUG mode
------------------------------------------- WORKERMAN --------------------------------------------
Workerman version:4.0.25          PHP version:7.1.25
-------------------------------------------- WORKERS ---------------------------------------------
proto   user            worker          listen                      processes    status
tcp     root            pusher          websocket://0.0.0.0:4236    2             [OK]
--------------------------------------------------------------------------------------------------
Press Ctrl+C to stop. Start success.
然後在43.65上也開啓第二台wsServer：
[root@localhost workerman]# /usr/local/php7125/bin/php -c /usr/local/php7125/lib/php.ini wsServer.php start
Workerman[wsServer.php] start in DEBUG mode
------------------------------------------- WORKERMAN --------------------------------------------
Workerman version:4.0.25          PHP version:7.1.25
-------------------------------------------- WORKERS ---------------------------------------------
proto   user            worker          listen                      processes    status
tcp     root            pusher          websocket://0.0.0.0:4236    2             [OK]
--------------------------------------------------------------------------------------------------
Press Ctrl+C to stop. Start success.
然後再開啓一臺httpServer(43.200)：
[root@localhost workerman]# /usr/local/php7125/bin/php -c /usr/local/php7125/lib/php/php.ini httpServer.php start
Workerman[httpServer.php] start in DEBUG mode
----------------------------------------- WORKERMAN -----------------------------------------
Workerman version:4.0.25          PHP version:7.1.25
------------------------------------------ WORKERS ------------------------------------------
proto   user            worker          listen                 processes    status
tcp     root            publisher       http://0.0.0.0:4237    1             [OK]
---------------------------------------------------------------------------------------------
Press Ctrl+C to stop. Start success.
httpServer脚本：

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

// 用来处理http请求，向任意客户端推送数据，需要传workerID和connectionID
$http_worker = new Worker('http://0.0.0.0:4237');
$http_worker->name = 'publisher';
$http_worker->onWorkerStart = function()
{
    Channel\Client::connect('192.168.43.170', 2206);
};
$http_worker->onMessage = function(TcpConnection $connection, $request)
{
    // 兼容workerman4.x
    if (!is_array($request)) {
        $_GET = $request->get();
    }
    $connection->send('ok');
    if(empty($_GET['content'])) return;
    // 是向某个worker进程中某个连接推送数据
    if(isset($_GET['to_worker_id']) && isset($_GET['to_connection_id']))
    {
        $event_name = $_GET['to_worker_id'];
        $to_connection_id = $_GET['to_connection_id'];
        $content = $_GET['content'];
        Channel\Client::publish($event_name, array(
            'to_connection_id' => $to_connection_id,
            'content'          => $content
        ));
    }
    // 是全局广播数据
    else
    {
        $event_name = '广播';
        $content = $_GET['content'];
        Channel\Client::publish($event_name, array(
            'content'          => $content
        ));
    }
};

Worker::runAll();

這裏httpServer負責發佈廣播或向某個進程中的某個websocket客戶端發送消息

然後再啓動第二台httpServer(43.109)：
[root@localhost workerman]# /usr/local/php7125/bin/php -c /usr/local/php7125/lib/php/php.ini httpServer.php start
Workerman[httpServer.php] start in DEBUG mode
----------------------------------------- WORKERMAN -----------------------------------------
Workerman version:4.0.25          PHP version:7.1.25
------------------------------------------ WORKERS ------------------------------------------
proto   user            worker          listen                 processes    status
tcp     root            publisher       http://0.0.0.0:4237    1             [OK]
---------------------------------------------------------------------------------------------
Press Ctrl+C to stop. Start success.
 
那麽接下來只要上面這幾臺服務器防火墻是放行對應端口的，則
可以同時開啓兩個websocket客戶端訪問，脚本分別是：
index.html:
<!doctype html>
<html>
<head>
    <script>
        ws1 = new WebSocket("ws://192.168.43.242:4236");
        ws1.onmessage = function(e) {
            alert("ws1收到服务端的消息：" + e.data);
        };
    </script>
</head>
<body>
    <div class="container">
        <h1>恭喜, 站点创建成功！</h1>
        <h3>这是默认index.html，本页面由系统自动生成</h3>
       <ul>
            <li>本页面在FTP根目录下的index.html</li>
            <li>您可以修改、删除或覆盖本页面</li>
            <li>FTP相关信息，请到“面板系统后台 > FTP” 查看</li>
        </ul>
    </div>
</body>
</html>

 

和index2.html:
<!doctype html>
<html>
<head>
    <script>
        ws2 = new WebSocket("ws://192.168.43.65:4236");
        ws2.onmessage = function(e) {
            alert("ws2收到服务端的消息：" + e.data);
        };
    </script>
</head>
<body>
    <div class="container">
        <h1>恭喜, 站点创建成功！</h1>
        <h3>这是默认index.html，本页面由系统自动生成</h3>
        <ul>
            <li>本页面在FTP根目录下的index.html</li>
            <li>您可以修改、删除或覆盖本页面</li>
            <li>FTP相关信息，请到“面板系统后台 > FTP” 查看</li>
        </ul>
    </div>
</body>
</html>

然後訪問如：

http://192.168.43.200:4237/?to_worker_id=0&to_connection_id=2&content=niaho
則是對43.200的http服務請求，對進程id為0且客戶端連接id為2的websocket用戶發消息，然後由於這個http服務將這個消息推送到channel中了，然websocket服務端有訂閲這個根據自己進程id訂閲的，正好兩個websocket服務端的進程id相同，且客戶端連接相同時候則有可能兩個websocket客戶端都收到消息，否則就只是某個websockt服務端旗下的指定客戶端收到。
而廣播推送則訪問如：
http://192.168.43.200:4237/?content=niaho
會出現目前兩個客戶端都接收的情況機會更大，因爲這兩個websocket的進程id由於不在一個服務器之中，所以默認情況下進程id都是0，每個workerman的第一個worker實例進程id都是從0開始的。
當然推送也是可以109的，我們部署了兩臺httpServer。
http://192.168.43.109:4237/?to_worker_id=0&to_connection_id=2&content=niaho
同理，廣播也可以是43.109