Channel分布式通讯组件 - 例子—集群推送
优质
小牛编辑
131浏览
2023-12-01
(要求Workerman版本>=3.3.0)
基于Worker的多进程(分布式集群)推送系统
start.php
<?php
use WorkermanWorker;
require_once __DIR__ . '/Workerman/Autoloader.php';
require_once __DIR__ . '/Channel/src/Server.php';
require_once __DIR__ . '/Channel/src/Client.php';
// 初始化一个Channel服务端
$channel_server = new ChannelServer('0.0.0.0', 2206);
// websocket服务端
$worker = new Worker('websocket://0.0.0.0:4236');
$worker->count=2;
$worker->name = 'pusher';
$worker->onWorkerStart = function($worker)
{
// Channel客户端连接到Channel服务端
ChannelClient::connect('127.0.0.1', 2206);
// 以自己的进程id为事件名称
$event_name = $worker->id;
// 订阅worker->id事件并注册事件处理函数
ChannelClient::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 existsn";
return;
}
$to_connection = $worker->connections[$to_connection_id];
$to_connection->send($message);
});
// 订阅广播事件
$event_name = '广播';
// 收到广播事件后向当前进程内所有客户端连接发送广播数据
ChannelClient::on($event_name, function($event_data)use($worker){
$message = $event_data['content'];
foreach($worker->connections as $connection)
{
$connection->send($message);
}
});
};
$worker->onConnect = function($connection)use($worker)
{
$msg = "workerID:{$worker->id} connectionID:{$connection->id} connectedn";
echo $msg;
$connection->send($msg);
};
// 用来处理http请求,向任意客户端推送数据,需要传workerID和connectionID
$http_worker = new Worker('http://0.0.0.0:4237');
$http_worker->name = 'publisher';
$http_worker->onWorkerStart = function()
{
ChannelClient::connect('127.0.0.1', 2206);
};
$http_worker->onMessage = function($connection, $data)
{
$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'];
ChannelClient::publish($event_name, array(
'to_connection_id' => $to_connection_id,
'content' => $content
));
}
// 是全局广播数据
else
{
$event_name = '广播';
$content = $_GET['content'];
ChannelClient::publish($event_name, array(
'content' => $content
));
}
};
Worker::runAll();
测试 (假设都是本机127.0.0.1运行)
1、运行服务端
php start.php start
Workerman[start.php] start in DEBUG mode
----------------------- WORKERMAN -----------------------------
Workerman version:3.2.7 PHP version:5.4.37
------------------------ WORKERS -------------------------------
user worker listen processes status
root ChannelServer frame://0.0.0.0:2206 1 [OK]
root pusher websocket://0.0.0.0:4236 2 [OK]
root publisher http://0.0.0.0:4237 1 [OK]
----------------------------------------------------------------
Press Ctrl-C to quit. Start success.
2、客户端连接服务端
打开chrome浏览器,按F12打开调试控制台,在Console一栏输入(或者把下面代码放入到html页面用js运行)
// 假设服务端ip为127.0.0.1,测试时请改成实际服务端ip
ws = new WebSocket("ws://127.0.0.1:4236");
ws.onmessage = function(e) {
alert("收到服务端的消息:" + e.data);
};
3、通过调用http接口推送
url访问 http://127.0.0.1:4237/?content={$content}
向所有客户端连接推送$content
数据
url访问 http://127.0.0.1:4237/?to_worker_id={$worker_id}&to_connection_id={$connection_id}&content={$content}
向某个worker进程中的某个客户端连接推送$content
数据
注意:测试时把127.0.0.1
{$worker_id}
{$connection_id}
和{$content}
换成实际值