如何基于Hyperf实现RabbitMQ+WebSocket消息推送

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

思路

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

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

WebSocket 服务

composer require hyperf/websocket-server

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

 SWOOLE_PROCESS,
    &＃39;servers&＃39; => [
        [
            &＃39;name&＃39; => &＃39;http&＃39;,
            &＃39;type&＃39; => Server::SERVER_HTTP,
            &＃39;host&＃39; => &＃39;0.0.0.0&＃39;,
            &＃39;port&＃39; => 11111,
            &＃39;sock_type&＃39; => SWOOLE_SOCK_TCP,
            &＃39;callbacks&＃39; => [
                SwooleEvent::ON_REQUEST => [Hyperf\HttpServer\Server::class, &＃39;onRequest&＃39;],
            ],
        ],
        [
            &＃39;name&＃39; => &＃39;ws&＃39;,
            &＃39;type&＃39; => Server::SERVER_WEBSOCKET,
            &＃39;host&＃39; => &＃39;0.0.0.0&＃39;,
            &＃39;port&＃39; => 12222,
            &＃39;sock_type&＃39; => SWOOLE_SOCK_TCP,
            &＃39;callbacks&＃39; => [
                SwooleEvent::ON_HAND_SHAKE => [Hyperf\WebSocketServer\Server::class, &＃39;onHandShake&＃39;],
                SwooleEvent::ON_MESSAGE => [Hyperf\WebSocketServer\Server::class, &＃39;onMessage&＃39;],
                SwooleEvent::ON_CLOSE => [Hyperf\WebSocketServer\Server::class, &＃39;onClose&＃39;],
            ],
        ],
    ],

WebSocket 服务器端代码示例

container->get(\Redis::class);
        //获取所有的客户端id
        $fdList = $redis->sMembers(&＃39;websocket_sjd_1&＃39;);
        //如果当前客户端在客户端集合中,就刷新
        if (in_array($frame->fd, $fdList)) {
            $redis->sAdd(&＃39;websocket_sjd_1&＃39;, $frame->fd);
            $redis->expire(&＃39;websocket_sjd_1&＃39;, 7200);
        }
        $server->push($frame->fd, &＃39;Recv: &＃39; . $frame->data);
    }
    /**
     * 客户端失去链接
     * @param Server $server
     * @param int $fd
     * @param int $reactorId
     */
    public function onClose(Server $server, int $fd, int $reactorId): void
    {
        //删掉客户端id
        $redis = $this->container->get(\Redis::class);
        //移除集合中指定的value
        $redis->sRem(&＃39;websocket_sjd_1&＃39;, $fd);
        var_dump(&＃39;closed&＃39;);
    }
    /**
     * 客户端链接
     * @param WebSocketServer $server
     * @param Request $request
     */
    public function onOpen(WebSocketServer $server, Request $request): void
    {
        //保存客户端id
        $redis = $this->container->get(\Redis::class);
        $res1 = $redis->sAdd(&＃39;websocket_sjd_1&＃39;, $request->fd);
        var_dump($res1);
        $res = $redis->expire(&＃39;websocket_sjd_1&＃39;, 7200);
        var_dump($res);
        $server->push($request->fd, &＃39;Opened&＃39;);
    }
}

WebSocket 前端代码

    function WebSocketTest() {
        if ("WebSocket" in window) {
            console.log("您的浏览器支持 WebSocket!");
            var num = 0
            // 打开一个 web socket
            var ws = new WebSocket("ws://127.0.0.1:12222");
            ws.Onopen= function () {
                // Web Socket 已连接上，使用 send() 方法发送数据
                //alert("数据发送中...");
                //ws.send("发送数据");
            };
            window.setInterval(function () { //每隔5秒钟发送一次心跳，避免websocket连接因超时而自动断开
                var ping = {"type": "ping"};
                ws.send(JSON.stringify(ping));
            }, 5000);
            ws.Onmessage= function (evt) {
                var d = JSON.parse(evt.data);
                console.log(d);
                if (d.code == 300) {
                    $(".address").text(d.address)
                }
                if (d.code == 200) {
                    var v = d.data
                    console.log(v);
                    num++
                    var str = `

                                    
${v.recordOutTime}
                                    
${v.userOutName}
                                    
${v.userOutNum}
                                    
${v.doorOutName}
                                
`
                    $(".tableHead").after(str)
                    if (num > 7) {
                        num--
                        $(".table .item:nth-last-child(1)").remove()
                    }
                }
            };
            ws.error = function (e) {
                console.log(e)
                alert(e)
            }
            ws.Onclose= function () {
                // 关闭 websocket
                alert("连接已关闭...");
            };
        } else {
            alert("您的浏览器不支持 WebSocket!");
        }
    }

AMQP 组件

composer require hyperf/amqp

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

 [
        &＃39;host&＃39; => &＃39;localhost&＃39;,
        &＃39;port&＃39; => 5672,
        &＃39;user&＃39; => &＃39;guest&＃39;,
        &＃39;password&＃39; => &＃39;guest&＃39;,
        &＃39;vhost&＃39; => &＃39;/&＃39;,
        &＃39;pool&＃39; => [
            &＃39;min_connections&＃39; => 1,
            &＃39;max_connections&＃39; => 10,
            &＃39;connect_timeout&＃39; => 10.0,
            &＃39;wait_timeout&＃39; => 3.0,
            &＃39;heartbeat&＃39; => -1,
        ],
        &＃39;params&＃39; => [
            &＃39;insist&＃39; => false,
            &＃39;login_method&＃39; => &＃39;AMQPLAIN&＃39;,
            &＃39;login_response&＃39; => null,
            &＃39;locale&＃39; => &＃39;en_US&＃39;,
            &＃39;connection_timeout&＃39; => 3.0,
            &＃39;read_write_timeout&＃39; => 6.0,
            &＃39;context&＃39; => null,
            &＃39;keepalive&＃39; => false,
            &＃39;heartbeat&＃39; => 3,
        ],
    ],
];

MQ 消费者代码

container->get(\Redis::class);
        $fdList=$redis->sMembers(&＃39;websocket_sjd_1&＃39;);
        $server=$this->container->get(ServerFactory::class)->getServer()->getServer();
        foreach($fdList as $key=>$v){
            if(!empty($v)){
                $server->push((int)$v, $data);
            }
        }
        return Result::ACK;
    }
}

控制器代码

    /**
     * test
     * @return array
     */
    public function test()
    {
        $data = array(
            &＃39;code&＃39; => 200,
            &＃39;data&＃39; => [
                &＃39;userOutName&＃39; => &＃39;ccflow&＃39;,
                &＃39;userOutNum&＃39; => &＃39;9999&＃39;,
                &＃39;recordOutTime&＃39; => date("Y-m-d H:i:s", time()),
                &＃39;doorOutName&＃39; => &＃39;教师公寓&＃39;,
            ]
        );
        $data = \GuzzleHttp\json_encode($data);
        $message = new DemoProducer($data);
        $producer = ApplicationContext::getContainer()->get(Producer::class);
        $result = $producer->produce($message);
        var_dump($result);
        $user = $this->request->input(&＃39;user&＃39;, &＃39;Hyperf&＃39;);
        $method = $this->request->getMethod();
        return [
            &＃39;method&＃39; => $method,
            &＃39;message&＃39; => "{$user}.",
        ];
    }

最终效果

推荐：《PHP教程》

以上就是如何基于Hyperf实现RabbitMQ+WebSocket消息推送的详细内容，更多请关注其它相关文章！