例子-集群推送 · 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}``` 換成實際值