当前位置: 首页 > 文档资料 > WorkerMan 3.x 手册 >

Channel分布式通讯组件 - 例子—集群推送

优质
小牛编辑
132浏览
2023-12-01

(要求Workerman版本>=3.3.0)

基于Worker的多进程(分布式集群)推送系统

start.php

  1. <?php
  2. use WorkermanWorker;
  3. require_once __DIR__ . '/Workerman/Autoloader.php';
  4. require_once __DIR__ . '/Channel/src/Server.php';
  5. require_once __DIR__ . '/Channel/src/Client.php';
  6. // 初始化一个Channel服务端
  7. $channel_server = new ChannelServer('0.0.0.0', 2206);
  8. // websocket服务端
  9. $worker = new Worker('websocket://0.0.0.0:4236');
  10. $worker->count=2;
  11. $worker->name = 'pusher';
  12. $worker->onWorkerStart = function($worker)
  13. {
  14. // Channel客户端连接到Channel服务端
  15. ChannelClient::connect('127.0.0.1', 2206);
  16. // 以自己的进程id为事件名称
  17. $event_name = $worker->id;
  18. // 订阅worker->id事件并注册事件处理函数
  19. ChannelClient::on($event_name, function($event_data)use($worker){
  20. $to_connection_id = $event_data['to_connection_id'];
  21. $message = $event_data['content'];
  22. if(!isset($worker->connections[$to_connection_id]))
  23. {
  24. echo "connection not existsn";
  25. return;
  26. }
  27. $to_connection = $worker->connections[$to_connection_id];
  28. $to_connection->send($message);
  29. });
  30. // 订阅广播事件
  31. $event_name = '广播';
  32. // 收到广播事件后向当前进程内所有客户端连接发送广播数据
  33. ChannelClient::on($event_name, function($event_data)use($worker){
  34. $message = $event_data['content'];
  35. foreach($worker->connections as $connection)
  36. {
  37. $connection->send($message);
  38. }
  39. });
  40. };
  41. $worker->onConnect = function($connection)use($worker)
  42. {
  43. $msg = "workerID:{$worker->id} connectionID:{$connection->id} connectedn";
  44. echo $msg;
  45. $connection->send($msg);
  46. };
  47. // 用来处理http请求,向任意客户端推送数据,需要传workerID和connectionID
  48. $http_worker = new Worker('http://0.0.0.0:4237');
  49. $http_worker->name = 'publisher';
  50. $http_worker->onWorkerStart = function()
  51. {
  52. ChannelClient::connect('127.0.0.1', 2206);
  53. };
  54. $http_worker->onMessage = function($connection, $data)
  55. {
  56. $connection->send('ok');
  57. if(empty($_GET['content'])) return;
  58. // 是向某个worker进程中某个连接推送数据
  59. if(isset($_GET['to_worker_id']) && isset($_GET['to_connection_id']))
  60. {
  61. $event_name = $_GET['to_worker_id'];
  62. $to_connection_id = $_GET['to_connection_id'];
  63. $content = $_GET['content'];
  64. ChannelClient::publish($event_name, array(
  65. 'to_connection_id' => $to_connection_id,
  66. 'content' => $content
  67. ));
  68. }
  69. // 是全局广播数据
  70. else
  71. {
  72. $event_name = '广播';
  73. $content = $_GET['content'];
  74. ChannelClient::publish($event_name, array(
  75. 'content' => $content
  76. ));
  77. }
  78. };
  79. Worker::runAll();

测试 (假设都是本机127.0.0.1运行)

1、运行服务端

  1. php start.php start
  2. Workerman[start.php] start in DEBUG mode
  3. ----------------------- WORKERMAN -----------------------------
  4. Workerman version:3.2.7 PHP version:5.4.37
  5. ------------------------ WORKERS -------------------------------
  6. user worker listen processes status
  7. root ChannelServer frame://0.0.0.0:2206 1 [OK]
  8. root pusher websocket://0.0.0.0:4236 2 [OK]
  9. root publisher http://0.0.0.0:4237 1 [OK]
  10. ----------------------------------------------------------------
  11. 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:4236");
  3. ws.onmessage = function(e) {
  4. alert("收到服务端的消息:" + e.data);
  5. };

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} 换成实际值