例子1
(要求Workerman版本>=3.3.0)
基于Worker的多進(jìn)程(分布式集群)推送系統(tǒng),集群群發(fā)、集群廣播。
start_channel.php
整個(gè)系統(tǒng)只能部署一個(gè)start_channel服務(wù)。假設(shè)運(yùn)行在192.168.1.1。
<?php
use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';
// 初始化一個(gè)Channel服務(wù)端
$channel_server = new Channel\Server('0.0.0.0', 2206);
Worker::runAll();
start_ws.php
整個(gè)系統(tǒng)可以部署多個(gè)start_ws服務(wù),假設(shè)運(yùn)行在 192.168.1.2和192.168.1.3 兩臺(tái)服務(wù)器上。
<?php
use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';
// websocket服務(wù)端
$worker = new Worker('websocket://0.0.0.0:4236');
$worker->count=2;
$worker->name = 'pusher';
$worker->onWorkerStart = function($worker)
{
// Channel客戶端連接到Channel服務(wù)端
Channel\Client::connect('192.168.1.1', 2206);
// 以自己的進(jìn)程id為事件名稱
$event_name = $worker->id;
// 訂閱worker->id事件并注冊(cè)事件處理函數(shù)
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 = '廣播';
// 收到廣播事件后向當(dāng)前進(jìn)程內(nèi)所有客戶端連接發(fā)送廣播數(shù)據(jù)
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();
start_http.php
整個(gè)系統(tǒng)可以部署多個(gè)start_http服務(wù),假設(shè)運(yùn)行在 192.168.1.4和192.168.1.5 兩臺(tái)服務(wù)器上。
<?php
use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';
// 用來處理http請(qǐng)求,向任意客戶端推送數(shù)據(jù),需要傳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.1.1', 2206);
};
$http_worker->onMessage = function(TcpConnection $connection, $request)
{
// 兼容workerman4.x 5.x
if (!is_array($request)) {
$get = $request->get();
}
$connection->send('ok');
if(empty($get['content'])) return;
// 是向某個(gè)worker進(jìn)程中某個(gè)連接推送數(shù)據(jù)
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
));
}
// 是全局廣播數(shù)據(jù)
else
{
$event_name = '廣播';
$content = $get['content'];
Channel\Client::publish($event_name, array(
'content' => $content
));
}
};
Worker::runAll();
測(cè)試
1、運(yùn)行個(gè)個(gè)服務(wù)器上的服務(wù)
2、客戶端連接服務(wù)端
打開chrome瀏覽器,按F12打開調(diào)試控制臺(tái),在Console一欄輸入(或者把下面代碼放入到html頁(yè)面用js運(yùn)行)
// 也可以連ws://192.168.1.3:4236
ws = new WebSocket("ws://192.168.1.2:4236");
ws.onmessage = function(e) {
alert("收到服務(wù)端的消息:" + e.data);
};
3、通過調(diào)用http接口推送
url訪問 http://192.168.1.4:4237/?content={$content}
或者 http://192.168.1.5:4237/?content={$content}
向所有客戶端連接推送$content
數(shù)據(jù)
url訪問 http://192.168.1.4:4237/?to_worker_id={$worker_id}&to_connection_id={$connection_id}&content={$content}
或者http://192.168.1.5:4237/?to_worker_id={$worker_id}&to_connection_id={$connection_id}&content={$content}
向某個(gè)worker進(jìn)程中的某個(gè)客戶端連接推送$content
數(shù)據(jù)
注意:測(cè)試時(shí)把 {$worker_id}
{$connection_id}
和{$content}
換成實(shí)際值