Hyperf WebSocket服务

Hyperf 提供了对 WebSocket Server 的封装，可基于 hyperf/websocket-server 组件快速搭建一个 WebSocket 应用。可用于即时通讯，如客服聊天系统，直播间聊天系统。这篇文章是我一点一点走出来的，全是干货，内容上更有衔接性，推荐大家去看。

安装

composer require hyperf/websocket-server

配置Server

修改 config/autoload/server.php，增加以下配置。

<?php

return [
    'servers' => [
        [
            'name' => 'ws',
            'type' => Server::SERVER_WEBSOCKET,
            'host' => '0.0.0.0',
            'port' => 9502,
            'sock_type' => SWOOLE_SOCK_TCP,
            'callbacks' => [
                Event::ON_HAND_SHAKE => [Hyperf\WebSocketServer\Server::class, 'onHandShake'],
                Event::ON_MESSAGE => [Hyperf\WebSocketServer\Server::class, 'onMessage'],
                Event::ON_CLOSE => [Hyperf\WebSocketServer\Server::class, 'onClose'],
            ],
        ],
    ],
];

配置路由

在 config/routes.php 文件内增加对应 ws 的 Server 的路由配置，这里的 ws 值取决于您在 config/autoload/server.php 内配置的 WebSocket Server 的 name 值。

<?php

Router::addServer('ws', function () {
    Router::get('/', 'App\Controller\WebSocketController');
});

配置中间件

在 config/autoload/middlewares.php 文件内增加对应 ws 的 Server 的全局中间件配置，这里的 ws 值取决于您在 config/autoload/server.php 内配置的 WebSocket Server 的 name 值。（非必须如果使用不到中间件可以不配置）

<?php

return [
    'ws' => [
        yourMiddleware::class
    ]
];

创建对应控制器（服务端）

<?php
declare(strict_types=1);

namespace App\Controller;

use Hyperf\Contract\OnCloseInterface;
use Hyperf\Contract\OnMessageInterface;
use Hyperf\Contract\OnOpenInterface;
use Hyperf\WebSocketServer\Constant\Opcode;
use Swoole\Http\Request;
use Swoole\Server;
use Swoole\Websocket\Frame;
use Swoole\WebSocket\Server as WebSocketServer;

class WebSocketController implements OnMessageInterface, OnOpenInterface, OnCloseInterface
{
    public function onMessage($server, Frame $frame): void
    {
        if($frame->opcode == Opcode::PING) {
            // 如果使用协程 Server，在判断是 PING 帧后，需要手动处理，返回 PONG 帧。
            // 异步风格 Server，可以直接通过 Swoole 配置处理，详情请见 https://wiki.swoole.com/#/websocket_server?id=open_websocket_ping_frame
            $server->push('', Opcode::PONG);
            return;
        }
        $server->push($frame->fd, 'Recv: ' . $frame->data);
    }

    public function onClose($server, int $fd, int $reactorId): void
    {
        var_dump('closed');
        $server->disconnect($fd);
    }

    public function onOpen($server, Request $request): void
    {
        $server->push($request->fd, 'Opened');
    }
}

创建 WebSocket Client （JS客户端）

您便可以通过各种 WebSocket Client 来进行连接和数据传输这里仅举例html方式

var wsServer = 'ws://127.0.0.1:9502';
var websocket = new WebSocket(wsServer);
websocket.onopen = function (evt) {
    console.log("Connected to WebSocket server.");
};

websocket.onclose = function (evt) {
    console.log("Disconnected");
};

websocket.onmessage = function (evt) {
    console.log('Retrieved data from server: ' + evt.data);
};

websocket.onerror = function (evt, e) {
    console.log('Error occured: ' + evt.data);
};

HTTP服务可以直接进行websocket连接

当我们同时监听了 HTTP Server 的 9501 端口和 WebSocket Server 的 9502 端口时， WebSocket Client 可以通过 9501 和 9502 两个端口连接 WebSocket Server，即连接 ws://0.0.0.0:9501 和 ws://0.0.0.0:9502 都可以成功。

因为 Swoole\WebSocket\Server 继承自 Swoole\Http\Server，可以使用 HTTP 触发所有 WebSocket 的推送，了解详情可查看 Swoole 文档 onRequest 回调部分。

如需关闭，可以修改 config/autoload/server.php 文件给 http 服务中增加 open_websocket_protocol 配置项。

<?php
return [
    // 这里省略了该文件的其它配置
    'servers' => [
        [
            'name' => 'http',
            'type' => Server::SERVER_HTTP,
            'host' => '0.0.0.0',
            'port' => 9501,
            'sock_type' => SWOOLE_SOCK_TCP,
            'callbacks' => [
                Event::ON_REQUEST => [Hyperf\HttpServer\Server::class, 'onRequest'],
            ],
            'settings' => [
                'open_websocket_protocol' => false,
            ]
        ],
    ]
];

HTTP服务中处理websocket(消息发送器)

当我们想在 HTTP 服务中，关闭 WebSocket 连接时，可以直接使用 Hyperf\WebSocketServer\Sender。

Sender 会判断 fd 是否被当前 Worker 所持有，如果是，则会直接发送数据，如果不是，则会通过 PipeMessage 发送给除自己外的所有 Worker，然后由其他 Worker 进行判断，如果是自己持有的 fd，就会发送对应数据到客户端。

Sender 支持 push 和 disconnect 两个 API，如下：

<?php

declare(strict_types=1);

namespace App\Controller;

use Hyperf\Di\Annotation\Inject;
use Hyperf\HttpServer\Annotation\AutoController;
use Hyperf\WebSocketServer\Sender;

#[AutoController]
class ServerController
{
    /**
     * @var Sender
     */
    #[Inject]
    protected $sender;

    public function close(int $fd)
    {
        go(function () use ($fd) {
            sleep(1);
            $this->sender->disconnect($fd);
        });

        return '';
    }

    public function send(int $fd)
    {
        $this->sender->push($fd, 'Hello World.');

        return '';
    }
}

连接上下文

WebSocket 服务的 onOpen, onMessage, onClose 回调并不在同一个协程下触发，因此不能直接使用协程上下文存储状态信息。WebSocket Server 组件提供了 连接级 的上下文，API 与协程上下文完全一样。

<?php
declare(strict_types=1);

namespace App\Controller;

use Hyperf\Contract\OnMessageInterface;
use Hyperf\Contract\OnOpenInterface;
use Hyperf\WebSocketServer\Context;
use Swoole\Http\Request;
use Swoole\Websocket\Frame;
use Swoole\WebSocket\Server as WebSocketServer;

class WebSocketController implements OnMessageInterface, OnOpenInterface
{
    public function onMessage($server, Frame $frame): void
    {
        $server->push($frame->fd, 'Username: ' . Context::get('username'));
    }

    public function onOpen($server, Request $request): void
    {
        Context::set('username', $request->cookie['username']);
    }
}

在 WebSocket 服务中处理 HTTP 请求(多端口监听)

Hyperf 支持监听多个端口，但因为 callbacks 中的对象直接从容器中获取，所以相同的 Hyperf\HttpServer\Server::class 会在容器中被覆盖。所以我们需要在依赖关系中，重新定义 Server，确保对象隔离。

WebSocket 和 TCP 等 Server 同理。

config/autoload/dependencies.php

<?php

return [
    'InnerHttp' => Hyperf\HttpServer\Server::class,
];

config/autoload/server.php

<?php
return [
    'servers' => [
        [
            'name' => 'http',
            'type' => Server::SERVER_HTTP,
            'host' => '0.0.0.0',
            'port' => 9501,
            'sock_type' => SWOOLE_SOCK_TCP,
            'callbacks' => [
                Event::ON_REQUEST => [Hyperf\HttpServer\Server::class, 'onRequest'],
            ],
        ],
        [
            'name' => 'innerHttp',
            'type' => Server::SERVER_HTTP,
            'host' => '0.0.0.0',
            'port' => 9502,
            'sock_type' => SWOOLE_SOCK_TCP,
            'callbacks' => [
                Event::ON_REQUEST => ['InnerHttp', 'onRequest'],
            ],
        ],
    ]
];

同时 路由文件，或者 注解 也需要指定对应的 server，如下：

路由文件 config/routes.php

<?php
Router::addServer('innerHttp', function () {
    Router::get('/', 'App\Controller\IndexController@index');
});

注解

<?php

declare(strict_types=1);

namespace App\Controller;

use Hyperf\HttpServer\Annotation\AutoController;

#[AutoController(server: "innerHttp")]
class IndexController
{
    public function index()
    {
        return 'Hello World.';
    }
}

在 WebSocket 服务中处理 HTTP 请求

我们除了上边可以将 HTTP 服务和 WebSocket 服务通过端口分开方式，也可以在 WebSocket 中监听 HTTP 请求。

因为 server.servers.*.callbacks 中的配置项，都是单例的，所以我们需要在 dependencies 中配置一个单独的实例。

<?php
return [
    'innerHttp' => Hyperf\HttpServer\Server::class,
];

然后修改我们的 WebSocket 服务中的 callbacks 配置，以下隐藏了不相干的配置

<?php

declare(strict_types=1);

use Hyperf\Server\Event;
use Hyperf\Server\Server;

return [
    'mode' => SWOOLE_BASE,
    'servers' => [
        [
            'name' => 'ws',
            'type' => Server::SERVER_WEBSOCKET,
            'host' => '0.0.0.0',
            'port' => 9502,
            'sock_type' => SWOOLE_SOCK_TCP,
            'callbacks' => [
                Event::ON_REQUEST => ['innerHttp', 'onRequest'],
                Event::ON_HAND_SHAKE => [Hyperf\WebSocketServer\Server::class, 'onHandShake'],
                Event::ON_MESSAGE => [Hyperf\WebSocketServer\Server::class, 'onMessage'],
                Event::ON_CLOSE => [Hyperf\WebSocketServer\Server::class, 'onClose'],
            ],
        ],
    ],
];

同理配置一下路由和注解

路由文件 config/routes.php



<?php
Router::addServer('innerHttp', function () {
    Router::get('/', 'App\Controller\IndexController@index');
});

注解



<?php

declare(strict_types=1);

namespace App\Controller;

use Hyperf\HttpServer\Annotation\AutoController;

#[AutoController(server: "innerHttp")]
class IndexController
{
    public function index()
    {
        return 'Hello World.';
    }
}

多server配置(未用到先记录，可能负载均衡会用)

# /etc/nginx/conf.d/ng_socketio.conf
# 多个 ws server
upstream io_nodes {
    server ws1:9502;
    server ws2:9502;
}
server {
  listen 9502;
  # server_name your.socket.io;
  location / {
    proxy_set_header Upgrade "websocket";
    proxy_set_header Connection "upgrade";
    # proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    # proxy_set_header Host $host;
    # proxy_http_version 1.1;
    # 转发到多个 ws server
    proxy_pass http://io_nodes;
  }
}

最终启动服务

php bin/hyperf.php start

如果启用多端口可以看见两个服务同时建立

[INFO] WebSocket Server listening at 0.0.0.0:9502
[INFO] HTTP Server listening at 0.0.0.0:9501

可参考成品：hyperf+webscoket搭建网站客服系统

相关文章: