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

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

思路

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

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

WebSocket 服务

composer require hyperf/websocket-server

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

<?phpreturn [    'mode' => SWOOLE_PROCESS,    'servers' => [        [            'name' => 'http',            'type' => Server::SERVER_HTTP,            'host' => '0.0.0.0',            'port' => 11111,            'sock_type' => SWOOLE_SOCK_TCP,            'callbacks' => [             <span style="color:transparent">~来1源gaodai#ma#com搞*代#码1网</span><abbr>搞代gaodaima码</abbr>   SwooleEvent::ON_REQUEST => [Hyperf\HttpServer\Server::class, 'onRequest'],            ],        ],        [            'name' => 'ws',            'type' => Server::SERVER_WEBSOCKET,            'host' => '0.0.0.0',            'port' => 12222,            'sock_type' => SWOOLE_SOCK_TCP,            'callbacks' => [                SwooleEvent::ON_HAND_SHAKE => [Hyperf\WebSocketServer\Server::class, 'onHandShake'],                SwooleEvent::ON_MESSAGE => [Hyperf\WebSocketServer\Server::class, 'onMessage'],                SwooleEvent::ON_CLOSE => [Hyperf\WebSocketServer\Server::class, 'onClose'],            ],        ],    ],

WebSocket 服务器端代码示例

<?phpdeclare(strict_types=1);/** * This file is part of Hyperf. * * @link     https://www.hyperf.io * @document https://doc.hyperf.io * @contact  [email protected] * @license  https://github.com/hyperf-cloud/hyperf/blob/master/LICENSE */namespace App\Controller;use Hyperf\Contract\OnCloseInterface;use Hyperf\Contract\OnMessageInterface;use Hyperf\Contract\OnOpenInterface;use Swoole\Http\Request;use Swoole\Server;use Swoole\Websocket\Frame;use Swoole\WebSocket\Server as WebSocketServer;class WebSocketController extends Controller implements OnMessageInterface, OnOpenInterface, OnCloseInterface{    /**     * 发送消息     * @param WebSocketServer $server     * @param Frame $frame     */    public function onMessage(WebSocketServer $server, Frame $frame): void    {        //心跳刷新缓存        $redis = $this->container->get(\Redis::class);        //获取所有的客户端id        $fdList = $redis->sMembers('websocket_sjd_1');        //如果当前客户端在客户端集合中,就刷新        if (in_array($frame->fd, $fdList)) {            $redis->sAdd('websocket_sjd_1', $frame->fd);            $redis->expire('websocket_sjd_1', 7200);        }        $server->push($frame->fd, 'Recv: ' . $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('websocket_sjd_1', $fd);        var_dump('closed');    }    /**     * 客户端链接     * @param WebSocketServer $server     * @param Request $request     */    public function onOpen(WebSocketServer $server, Request $request): void    {        //保存客户端id        $redis = $this->container->get(\Redis::class);        $res1 = $redis->sAdd('websocket_sjd_1', $request->fd);        var_dump($res1);        $res = $redis->expire('websocket_sjd_1', 7200);        var_dump($res);        $server->push($request->fd, 'Opened');    }}

喜欢 (0)赏

[搞代码]

分享 (0)