基于 Hyperf+ WebSocket +RabbitMQ 实现的一个简单大屏幕的消息推送。

思路

利用 WebSocket 协议让客户端和服务器端保持有状态的长链接，

保存链接上来的客户端 id。订阅发布者发布的消息针对已保存的客户端 id 进行广播消息。

WebSocket 服务

composer require hyperf/websocket-server

配置文件 [config/autoload/server.php]

1. <?php 
2.  
3. return [ 
4.  
5.     'mode' => SWOOLE_PROCESS, 
6.  
7.     'servers' => [ 
8.  
9.         [ 
10.  
11.             'name' => 'http', 
12.  
13.             'type' => Server::SERVER_HTTP, 
14.  
15.             'host' => '0.0.0.0', 
16.  
17.             'port' => 11111, 
18.  
19.             'sock_type' => SWOOLE_SOCK_TCP, 
20.  
21.             'callbacks' => [ 
22.  
23.                 SwooleEvent::ON_REQUEST => [Hyperf\HttpServer\Server::class, 'onRequest'], 
24.  
25.             ], 
26.  
27.         ], 
28.  
29.         [ 
30.  
31.             'name' => 'ws', 
32.  
33.             'type' => Server::SERVER_WEBSOCKET, 
34.  
35.             'host' => '0.0.0.0', 
36.  
37.             'port' => 12222, 
38.  
39.             'sock_type' => SWOOLE_SOCK_TCP, 
40.  
41.             'callbacks' => [ 
42.  
43.                 SwooleEvent::ON_HAND_SHAKE => [Hyperf\WebSocketServer\Server::class, 'onHandShake'], 
44.  
45.                 SwooleEvent::ON_MESSAGE => [Hyperf\WebSocketServer\Server::class, 'onMessage'], 
46.  
47.                 SwooleEvent::ON_CLOSE => [Hyperf\WebSocketServer\Server::class, 'onClose'], 
48.  
49.             ], 
50.  
51.         ], 
52.  
53.     ], 

WebSocket 服务器端代码示例：

1. <?php 
2.  
3. declare(strict_types=1); 
4.  
5. /** 
6.  
7.  * This file is part of Hyperf. 
8.  
9.  * 
10.  
11.  * @link     https://www.hyperf.io 
12.  
13.  * @document https://doc.hyperf.io 
14.  
15.  * @contact  group@hyperf.io 
16.  
17.  * @license  https://github.com/hyperf-cloud/hyperf/blob/master/LICENSE 
18.  
19.  */ 
20.  
21. namespace App\Controller; 
22.  
23. use Hyperf\Contract\OnCloseInterface; 
24.  
25. use Hyperf\Contract\OnMessageInterface; 
26.  
27. use Hyperf\Contract\OnOpenInterface; 
28.  
29. use Swoole\Http\Request; 
30.  
31. use Swoole\Server; 
32.  
33. use Swoole\Websocket\Frame; 
34.  
35. use Swoole\WebSocket\Server as WebSocketServer; 
36.  
37. class WebSocketController extends Controller implements OnMessageInterface, OnOpenInterface, OnCloseInterface 
38.  
39. { 
40.  
41.     /** 
42.  
43.      * 发送消息 
44.  
45.      * @param WebSocketServer $server 
46.  
47.      * @param Frame $frame 
48.  
49.      */ 
50.  
51.     public function onMessage(WebSocketServer $server, Frame $frame): void 
52.  
53.     { 
54.  
55.         //心跳刷新缓存 
56.  
57.         $redis = $this->container->get(\Redis::class); 
58.  
59.         //获取所有的客户端id 
60.  
61.         $fdList = $redis->sMembers('websocket_sjd_1'); 
62.  
63.         //如果当前客户端在客户端集合中,就刷新 
64.  
65.         if (in_array($frame->fd, $fdList)) { 
66.  
67.             $redis->sAdd('websocket_sjd_1', $frame->fd); 
68.  
69.             $redis->expire('websocket_sjd_1', 7200); 
70.  
71.         } 
72.  
73.         $server->push($frame->fd, 'Recv: ' . $frame->data); 
74.  
75.     } 
76.  
77.     /** 
78.  
79.      * 客户端失去链接 
80.  
81.      * @param Server $server 
82.  
83.      * @param int $fd 
84.  
85.      * @param int $reactorId 
86.  
87.      */ 
88.  
89.     public function onClose(Server $server, int $fd, int $reactorId): void 
90.  
91.     { 
92.  
93.         //删掉客户端id 
94.  
95.         $redis = $this->container->get(\Redis::class); 
96.  
97.         //移除集合中指定的value 
98.  
99.         $redis->sRem('websocket_sjd_1', $fd); 
100.  
101.         var_dump('closed'); 
102.  
103.     } 
104.  
105.     /** 
106.  
107.      * 客户端链接 
108.  
109.      * @param WebSocketServer $server 
110.  
111.      * @param Request $request 
112.  
113.      */ 
114.  
115.     public function onOpen(WebSocketServer $server, Request $request): void 
116.  
117.     { 
118.  
119.         //保存客户端id 
120.  
121.         $redis = $this->container->get(\Redis::class); 
122.  
123.         $res1 = $redis->sAdd('websocket_sjd_1', $request->fd); 
124.  
125.         var_dump($res1); 
126.  
127.         $res = $redis->expire('websocket_sjd_1', 7200); 
128.  
129.         var_dump($res); 
130.  
131.         $server->push($request->fd, 'Opened'); 
132.  
133.     } 
134.  
135. } 

WebSocket 前端代码：

1. function WebSocketTest() { 
2.  
3.         if ("WebSocket" in window) { 
4.  
5.             console.log("您的浏览器支持 WebSocket!"); 
6.  
7.             var num = 0 
8.  
9.             // 打开一个 web socket 
10.  
11.             var ws = new WebSocket("ws://127.0.0.1:12222"); 
12.  
13.             ws.onopen = function () { 
14.  
15.                 // Web Socket 已连接上，使用 send() 方法发送数据 
16.  
17.                 //alert("数据发送中..."); 
18.  
19.                 //ws.send("发送数据"); 
20.  
21.             }; 
22.  
23.             window.setInterval(function () { //每隔5秒钟发送一次心跳，避免websocket连接因超时而自动断开 
24.  
25.                 var ping = {"type": "ping"}; 
26.  
27.                 ws.send(JSON.stringify(ping)); 
28.  
29.             }, 5000); 
30.  
31.             ws.onmessage = function (evt) { 
32.  
33.                 var d = JSON.parse(evt.data); 
34.  
35.                 console.log(d); 
36.  
37.                 if (d.code == 300) { 
38.  
39.                     $(".address").text(d.address) 
40.  
41.                 } 
42.  
43.                 if (d.code == 200) { 
44.  
45.                     var v = d.data 
46.  
47.                     console.log(v); 
48.  
49.                     num++ 
50.  
51.                     var str = `<div class="item"> 
52.  
53.                                     <p>${v.recordOutTime}</p> 
54.  
55.                                     <p>${v.userOutName}</p> 
56.  
57.                                     <p>${v.userOutNum}</p> 
58.  
59.                                     <p>${v.doorOutName}</p> 
60.  
61.                                 </div>` 
62.  
63.                     $(".tableHead").after(str) 
64.  
65.                     if (num > 7) { 
66.  
67.                         num-- 
68.  
69.                         $(".table .item:nth-last-child(1)").remove() 
70.  
71.                     } 
72.  
73.                 } 
74.  
75.             }; 
76.  
77.             ws.error = function (e) { 
78.  
79.                 console.log(e) 
80.  
81.                 alert(e) 
82.  
83.             } 
84.  
85.             ws.onclose = function () { 
86.  
87.                 // 关闭 websocket 
88.  
89.                 alert("连接已关闭..."); 
90.  
91.             }; 
92.  
93.         } else { 
94.  
95.             alert("您的浏览器不支持 WebSocket!"); 
96.  
97.         } 
98.  
99.     } 

AMQP 组件

composer require hyperf/amqp

配置文件 [config/autoload/amqp.php]

1. <?php 
2.  
3. return [ 
4.  
5.     'default' => [ 
6.  
7.         'host' => 'localhost', 
8.  
9.         'port' => 5672, 
10.  
11.         'user' => 'guest', 
12.  
13.         'password' => 'guest', 
14.  
15.         'vhost' => '/', 
16.  
17.         'pool' => [ 
18.  
19.             'min_connections' => 1, 
20.  
21.             'max_connections' => 10, 
22.  
23.             'connect_timeout' => 10.0, 
24.  
25.             'wait_timeout' => 3.0, 
26.  
27.             'heartbeat' => -1, 
28.  
29.         ], 
30.  
31.         'params' => [ 
32.  
33.             'insist' => false, 
34.  
35.             'login_method' => 'AMQPLAIN', 
36.  
37.             'login_response' => null, 
38.  
39.             'locale' => 'en_US', 
40.  
41.             'connection_timeout' => 3.0, 
42.  
43.             'read_write_timeout' => 6.0, 
44.  
45.             'context' => null, 
46.  
47.             'keepalive' => false, 
48.  
49.             'heartbeat' => 3, 
50.  
51.         ], 
52.  
53.     ], 
54.  
55. ]; 

MQ 消费者代码

1. <?php 
2.  
3. declare(strict_types=1); 
4.  
5. namespace App\Amqp\Consumer; 
6.  
7. use Hyperf\Amqp\Annotation\Consumer; 
8.  
9. use Hyperf\Amqp\Message\ConsumerMessage; 
10.  
11. use Hyperf\Amqp\Result; 
12.  
13. use Hyperf\Server\Server; 
14.  
15. use Hyperf\Server\ServerFactory; 
16.  
17. /** 
18.  
19.  * @Consumer(exchange="hyperf", routingKey="hyperf", queue="hyperf", nums=1) 
20.  
21.  */ 
22.  
23. class DemoConsumer extends ConsumerMessage 
24.  
25. { 
26.  
27.     /** 
28.  
29.      * rabbmitMQ消费端代码 
30.  
31.      * @param $data 
32.  
33.      * @return string 
34.  
35.      */ 
36.  
37.     public function consume($data): string 
38.  
39.     { 
40.  
41.         print_r($data); 
42.  
43.         //获取集合中所有的value 
44.  
45.         $redis = $this->container->get(\Redis::class); 
46.  
47.         $fdList=$redis->sMembers('websocket_sjd_1'); 
48.  
49.         $server=$this->container->get(ServerFactory::class)->getServer()->getServer(); 
50.  
51.         foreach($fdList as $key=>$v){ 
52.  
53.             if(!emptyempty($v)){ 
54.  
55.                 $server->push((int)$v, $data); 
56.  
57.             } 
58.  
59.         } 
60.  
61.         return Result::ACK; 
62.  
63.     } 
64.  
65. } 

控制器代码

1. /** 
2.  
3.    * test 
4.  
5.    * @return array 
6.  
7.    */ 
8.  
9.   public function test() 
10.  
11.   { 
12.  
13.       $data = array( 
14.  
15.           'code' => 200, 
16.  
17.           'data' => [ 
18.  
19.               'userOutName' => 'ccflow', 
20.  
21.               'userOutNum' => '9999', 
22.  
23.               'recordOutTime' => date("Y-m-d H:i:s", time()), 
24.  
25.               'doorOutName' => '教师公寓', 
26.  
27.           ] 
28.  
29.       ); 
30.  
31.       $data = \GuzzleHttp\json_encode($data); 
32.  
33.       $message = new DemoProducer($data); 
34.  
35.       $producer = ApplicationContext::getContainer()->get(Producer::class); 
36.  
37.       $result = $producer->produce($message); 
38.  
39.       var_dump($result); 
40.  
41.       $user = $this->request->input('user', 'Hyperf'); 
42.  
43.       $method = $this->request->getMethod(); 
44.  
45.       return [ 
46.  
47.           'method' => $method, 
48.  
49.           'message' => "{$user}.", 
50.  
51.       ]; 
52.  
53.   } 

最终效果

Tags: Hyperf RabbitMQ WebSocket

分享到：