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

爱被打了一巴掌 2023-05-22 06:45 18阅读 0赞

## 思路 ##

利用 WebSocket 协议让客户端和服务器端保持有状态的长链接，保存链接上来的客户端 id。订阅发布者发布的消息针对已保存的客户端 id 进行广播消息。

我的官方群[点击此处][Link 1]。

## WebSocket 服务 ##

composer require hyperf/websocket-server

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

<?php
    
    return [
        'mode' => SWOOLE_PROCESS,
        'servers' => [
            [
                'name' => 'http',
                'type' => Server::SERVER_HTTP,
                'host' => '0.0.0.0',
                'port' => 11111,
                'sock_type' => SWOOLE_SOCK_TCP,
                'callbacks' => [
                    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 服务器端代码示例 ##

<?php
    
    declare(strict_types=1);
    /**
     * This file is part of Hyperf.
     *
     * @link     https://www.hyperf.io
     * @document https://doc.hyperf.io
     * @contact  group@hyperf.io
     * @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');
    
        }
    }

### WebSocket 前端代码 ###

## AMQP 组件 ##

composer require hyperf/amqp

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

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

## MQ 消费者代码 ##

<?php
    declare(strict_types=1);
    
    namespace App\Amqp\Consumer;
    
    use Hyperf\Amqp\Annotation\Consumer;
    use Hyperf\Amqp\Message\ConsumerMessage;
    use Hyperf\Amqp\Result;
    use Hyperf\Server\Server;
    use Hyperf\Server\ServerFactory;
    
    /**
     * @Consumer(exchange="hyperf", routingKey="hyperf", queue="hyperf", nums=1)
     */
    class DemoConsumer extends ConsumerMessage
    {
        /**
         * rabbmitMQ消费端代码
         * @param $data
         * @return string
         */
        public function consume($data): string
        {
            print_r($data);
    
            //获取集合中所有的value
            $redis = $this->container->get(\Redis::class);
            $fdList=$redis->sMembers('websocket_sjd_1');
    
            $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(
                'code' => 200,
                'data' => [
                    'userOutName' => 'ccflow',
                    'userOutNum' => '9999',
                    'recordOutTime' => date("Y-m-d H:i:s", time()),
                    'doorOutName' => '教师公寓',
                ]
            );
            $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('user', 'Hyperf');
            $method = $this->request->getMethod();
    
            return [
                'method' => $method,
                'message' => "{$user}.",
            ];
        }

## 最终效果 ##

![format_png][]

更多学习内容请访问：

[腾讯T3-T4标准精品PHP架构师教程目录大全，只要你看完保证薪资上升一个台阶（持续更新）][T3-T4_PHP]

**以上内容希望帮助到大家，**很多PHPer在进阶的时候总会遇到一些问题和瓶颈，业务代码写多了没有方向感，不知道该从那里入手去提升，对此我整理了一些资料，包括但不限于：分布式架构、高可扩展、高性能、高并发、服务器性能调优、TP6，laravel，YII2，Redis，Swoole、Swoft、Kafka、Mysql优化、shell脚本、Docker、微服务、Nginx等多个知识点高级进阶干货需要的可以免费分享给大家**，需要的可以加入**我的官方群[点击此处][Link 1]。

[Link 1]: https://link.zhihu.com/?target=https%3A//jq.qq.com/%3F_wv%3D1027%26k%3D5kN6y8c
[format_png]: /images/20230521/12c1a878870d4d4d95d1b3ff343da5c0.png
[T3-T4_PHP]: https://zhuanlan.zhihu.com/p/102278033