例子1

(要求Workerman版本>=3.3.0)

基于Worker的多进程分组推送系统

  1.     use Workerman\Worker;
  2.     use Workerman\Connection\TcpConnection;
  3.     require_once __DIR__ . '/vendor/autoload.php';
  5.     $channel_server = new Channel\Server('0.0.0.0', 2206);
  7.     $worker = new Worker('websocket://0.0.0.0:1234');
  8.     $worker->count = 8;
  9.     // 全局群组到连接的映射数组
  10.     $group_con_map = array();
  11.     $worker->onWorkerStart = function(){
  12.         // Channel客户端连接到Channel服务端
  13.         Channel\Client::connect('127.0.0.1', 2206);
  15.         // 监听全局分组发送消息事件
  16.         Channel\Client::on('send_to_group', function($event_data){
  17.             $group_id = $event_data['group_id'];
  18.             $message = $event_data['message'];
  19.             global $group_con_map;
  20.             var_dump(array_keys($group_con_map));
  21.             if (isset($group_con_map[$group_id])) {
  22.                 foreach ($group_con_map[$group_id] as $con) {
  23.                     $con->send($message);
  24.                 }
  25.             }
  26.         });
  27.     };
  28.     $worker->onMessage = function(TcpConnection $con, $data){
  29.         // 加入群组消息{"cmd":"add_group", "group_id":"123"}
  30.         // 或者 群发消息{"cmd":"send_to_group", "group_id":"123", "message":"这个是消息"}
  31.         $data = json_decode($data, true);
  32.         var_dump($data);
  33.         $cmd = $data['cmd'];
  34.         $group_id = $data['group_id'];
  35.         switch($cmd) {
  36.             // 连接加入群组
  37.             case "add_group":
  38.                 global $group_con_map;
  39.                 // 将连接加入到对应的群组数组里
  40.                 $group_con_map[$group_id][$con->id] = $con;
  41.                 // 记录这个连接加入了哪些群组，方便在onclose的时候清理group_con_map对应群组的数据
  42.                 $con->group_id = isset($con->group_id) ? $con->group_id : array();
  43.                 $con->group_id[$group_id] = $group_id;
  44.                 break;
  45.             // 群发消息给群组
  46.             case "send_to_group":
  47.                 // Channel\Client给所有服务器的所有进程广播分组发送消息事件
  48.                 Channel\Client::publish('send_to_group', array(
  49.                     'group_id'=>$group_id,
  50.                     'message'=>$data['message']
  51.                 ));
  52.                 break;
  53.         }
  54.     };
  55.     // 这里很重要，连接关闭时把连接从全局群组数据中删除，避免内存泄漏
  56.     $worker->onClose = function(TcpConnection $con){
  57.         global $group_con_map;
  58.         // 遍历连接加入的所有群组，从group_con_map删除对应的数据
  59.         if (isset($con->group_id)) {
  60.             foreach ($con->group_id as $group_id) {
  61.                 unset($group_con_map[$group_id][$con->id]);
  62.             }
  63.             if (empty($group_con_map[$group_id])) {
  64.                 unset($group_con_map[$group_id]);
  65.             }
  66.         }
  67.     };
  69.     Worker::runAll();

测试 （假设都是本机127.0.0.1运行）

1、运行服务端

  1. php start.php start
  2. Workerman[del.php] start in DEBUG mode
  3. ----------------------- WORKERMAN -----------------------------
  4. Workerman version:3.4.2          PHP version:7.1.3
  5. ------------------------ WORKERS -------------------------------
  6. user          worker         listen                    processes status
  7. liliang       ChannelServer  frame://0.0.0.0:2206       1         [OK] 
  8. liliang       none           websocket://0.0.0.0:1234   12        [OK] 
  9. ----------------------------------------------------------------
  10. Press Ctrl-C to quit. Start success.

2、客户端连接服务端

打开chrome浏览器，按F12打开调试控制台，在Console一栏输入(或者把下面代码放入到html页面用js运行)

  1. // 假设服务端ip为127.0.0.1，测试时请改成实际服务端ip
  2. ws = new WebSocket('ws://127.0.0.1:1234');
  3. ws.onmessage = function(data){console.log(data.data)};
  4. ws.onopen = function() {
  5.     ws.send('{"cmd":"add_group", "group_id":"123"}');
  6.     ws.send('{"cmd":"send_to_group", "group_id":"123", "message":"这个是消息"}');
  7. };