例子-集群推送 · swoole-worker

# 例子-集群推送 **`(要求Workerman版本>=3.3.0)`** 基于Worker的多进程(分布式集群)推送系统，集群群发、集群广播。 start.php ```php <?php use Workerman\Worker; require_once __DIR__ . '/Workerman/Autoloader.php'; require_once __DIR__ . '/Channel/src/Server.php'; require_once __DIR__ . '/Channel/src/Client.php'; // 初始化一个Channel服务端 $channel_server = new Channel\Server('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服务端 Channel\Client::connect('127.0.0.1', 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($connection)use($worker) { $msg = "workerID:{$worker->id} connectionID:{$connection->id} connected\n"; 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() { Channel\Client::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']; 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(); ``` ## 测试（假设都是本机127.0.0.1运行） 1、运行服务端 ```php 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运行) ```php // 假设服务端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}```](http://127.0.0.1:4237/?content=%7B%24content%7D%60%60%60) 向所有客户端连接推送```$content```数据 url访问 ```[http://127.0.0.1:4237/?to\_worker\_id={$worker\_id}&to\_connection\_id={$connection\_id}&content={$content}```向某个worker进程中的某个客户端连接推送```$content```数据](http://127.0.0.1:4237/?to_worker_id=%7B%24worker_id%7D&to_connection_id=%7B%24connection_id%7D&content=%7B%24content%7D%60%60%60%E5%90%91%E6%9F%90%E4%B8%AAworker%E8%BF%9B%E7%A8%8B%E4%B8%AD%E7%9A%84%E6%9F%90%E4%B8%AA%E5%AE%A2%E6%88%B7%E7%AB%AF%E8%BF%9E%E6%8E%A5%E6%8E%A8%E9%80%81%60%60%60%24content%60%60%60%E6%95%B0%E6%8D%AE) 注意：测试时把```127.0.0.1````{$worker_id}````{$connection\_id}``` 和```{$content}``` 换成实际值