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

workerman常用组件-channel分佈式通訊組件-channelClient-例子-分組發送（聊天組應用場景）

发布时间：2021-12-19 17:56:53编辑：雪饮阅读（）

 (要求Workerman版本>=3.3.0)

基于Worker的多进程分组推送系统
基於上篇(http://www.gaojiupan.cn/manshenghuo/chengxurensheng/4596.html)這篇就簡單多了
不需要5台設備那麽奢侈了。
這裏爲了模擬效果，雖然至少需要兩個服務器，這裏爲了方便都放在同一臺服務器上面了，以不同的端口來區分。
本次主要是模擬消息群發，也就是常見的聊天組功能了。
則websocket服務端連同channel服務端的實現如：
<?php
use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';

$channel_server = new Channel\Server('0.0.0.0', 2206);

$worker = new Worker('websocket://0.0.0.0:1234');
$worker->count = 8;
// 全局群组到连接的映射数组
$group_con_map = array();
$worker->onWorkerStart = function(){
    // Channel客户端连接到Channel服务端
    Channel\Client::connect('127.0.0.1', 2206);

    // 监听全局分组发送消息事件
    Channel\Client::on('send_to_group', function($event_data){
        $group_id = $event_data['group_id'];
        $message = $event_data['message'];
        global $group_con_map;
        var_dump(array_keys($group_con_map));
        if (isset($group_con_map[$group_id])) {
            foreach ($group_con_map[$group_id] as $con) {
                $con->send($message);
            }
        }
    });
};
$worker->onMessage = function(TcpConnection $con, $data){
    // 加入群组消息{"cmd":"add_group", "group_id":"123"}
    // 或者 群发消息{"cmd":"send_to_group", "group_id":"123", "message":"这个是消息"}
    $data = json_decode($data, true);
    var_dump($data);
    $cmd = $data['cmd'];
    $group_id = $data['group_id'];
    switch($cmd) {
        // 连接加入群组
        case "add_group":
            global $group_con_map;
            // 将连接加入到对应的群组数组里
            $group_con_map[$group_id][$con->id] = $con;
            // 记录这个连接加入了哪些群组，方便在onclose的时候清理group_con_map对应群组的数据
            $con->group_id = isset($con->group_id) ? $con->group_id : array();
            $con->group_id[$group_id] = $group_id;
            break;
        // 群发消息给群组
        case "send_to_group":
            // Channel\Client给所有服务器的所有进程广播分组发送消息事件
            Channel\Client::publish('send_to_group', array(
                'group_id'=>$group_id,
                'message'=>$data['message']
            ));
            break;
    }
};
// 这里很重要，连接关闭时把连接从全局群组数据中删除，避免内存泄漏
$worker->onClose = function(TcpConnection $con){
    global $group_con_map;
    // 遍历连接加入的所有群组，从group_con_map删除对应的数据
    if (isset($con->group_id)) {
        foreach ($con->group_id as $group_id) {
            unset($group_con_map[$group_id][$con->id]);
        }
        if (empty($group_con_map[$group_id])) {
            unset($group_con_map[$group_id]);
        }
    }
};

Worker::runAll();
運行該服務端脚本並有websocket客戶端連接並發送了一些消息時：
[root@localhost www.fpm.com]# /usr/local/php734/bin/php -c /usr/local/php734/lib/php/php.ini allServer.php start
Workerman[allServer.php] start in DEBUG mode
-------------------------------------------- WORKERMAN --------------------------------------------
Workerman version:4.0.22          PHP version:7.3.4
--------------------------------------------- WORKERS ---------------------------------------------
proto   user            worker           listen                      processes    status
tcp     root            ChannelServer    frame://0.0.0.0:2206        1             [OK]
tcp     root            none             websocket://0.0.0.0:1234    8             [OK]
---------------------------------------------------------------------------------------------------
Press Ctrl+C to stop. Start success.
array(2) {
 ["cmd"]=>
 string(9) "add_group"
 ["group_id"]=>
 string(3) "123"
}
array(3) {
 ["cmd"]=>
 string(13) "send_to_group"
 ["group_id"]=>
 string(3) "123"
 ["message"]=>
 string(15) "这个是消息"
}
array(0) {
array(0) {
}
}
array(0) {
}
array(0) {
}
array(0) {
}
array(0) {
}
array(0) {
}
array(1) {
 [0]=>
 int(123)
}
array(2) {
 ["cmd"]=>
 string(9) "add_group"
 ["group_id"]=>
 string(3) "123"
}
array(3) {
 ["cmd"]=>
 string(13) "send_to_group"
 ["group_id"]=>
 string(3) "123"
 ["message"]=>
 string(15) "这个是消息"
}
array(0) {
}
array(0) {
}
array(0) {
}
array(0) {
}
array(0) {
array(0) {
}
}
array(1) {
 [0]=>
 int(123)
}
array(0) {
}
 
則websocket客戶端就很簡單了
<script>
    ws = new WebSocket('ws://192.168.43.170:1234');
    ws.onmessage = function(data){console.log(data.data)};
    ws.onopen = function() {
        ws.send('{"cmd":"add_group", "group_id":"123"}');
        ws.send('{"cmd":"send_to_group", "group_id":"123", "message":"这个是消息"}');
    };
</script>