hyperf-message-push/hyperf-skeleton/app/Controller/WebSocketController.php

<?php
declare(strict_types=1);

namespace App\Controller;

use Hyperf\Contract\OnCloseInterface;
use Hyperf\Contract\OnMessageInterface;
use Hyperf\Contract\OnOpenInterface;
use Swoole\Http\Request;
use Swoole\Websocket\Frame;

class WebSocketController extends AbstractController implements OnMessageInterface, OnOpenInterface, OnCloseInterface
{
    /**
     * 发送消息
     * @param $server
     * @param Frame $frame
     */
    public function onMessage($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);
        }

        // push消息给客户端
        $server->push($frame->fd, json_encode(['Recv' => $frame->data]));
    }

    /**
     * 客户端失去连接
     * @param $server
     * @param int $fd
     * @param int $reactorId
     */
    public function onClose($server, int $fd, int $reactorId): void
    {
        $redis = $this->container->get(\Redis::class);
        // 移除集合中指定的value
        $redis->sRem('websocket_sjd_1', $fd);
        echo 'closed';
    }

    /**
     * 客户端连接开启
     * @param $server
     * @param Request $request
     */
    public function onOpen($server, Request $request): void
    {
        $redis = $this->container->get(\Redis::class);
        // 保存客户端id
        $resAdd = $redis->sAdd('websocket_sjd_1', $request->fd);
        var_dump($resAdd);

        $resExp = $redis->expire('websocket_sjd_1', 7200);
        var_dump($resExp);

        // push消息给客户端
        $server->push($request->fd, json_encode(['Status' => 'Opened']));
    }
}