发表新帖

Swoole\Coroutine\Redis 在swoole中已移除，官方说的是使用hook方式，我的服务启动后websocket客户端无法连接了？什么原因啊？大佬帮忙看下！！！

```
<?php

namespace Layman\LaravelWebsocket\Server;

use co;
use Illuminate\Support\Facades\Log;
use Layman\LaravelWebsocket\Models\WebSocketMessage;
use Layman\LaravelWebsocket\Support\ConnectionManager;
use Layman\LaravelWebsocket\Support\DatabasePersistence;
use Layman\LaravelWebsocket\Support\Heartbeat;
use Layman\LaravelWebsocket\Support\MessageDispatcher;
use Layman\LaravelWebsocket\Support\MessageFormatter;
use Layman\LaravelWebsocket\Support\RedisPersistence;
use Predis\Client;
use Predis\ClientInterface;
use Predis\Consumer\Push\PushResponseInterface;
use Swoole\Http\Request;
use Swoole\WebSocket\Server;

class WebSocketServer
{

protected Server $server;
    protected ConnectionManager $connections;
    protected Heartbeat $heartbeat;
    protected MessageDispatcher $dispatcher;
    protected DatabasePersistence $databasePersistence;
    protected RedisPersistence $redisPersistence;
    protected array $config;

public function __construct()
    {
        $this->config = config('websocket');
        $this->server = new Server($this->config['host'], $this->config['port'], $this->config['model']);
        if ($this->config['model'] == SWOOLE_PROCESS) {
            $this->server->set([
                'worker_num' => $this->config['worker_num'],
            ]);
        }
        $this->connections         = new ConnectionManager();
        $this->heartbeat           = new Heartbeat($this->server, $this->connections, $this->config['heartbeat_interval'], $this->config['heartbeat_timeout']);
        $this->dispatcher          = new MessageDispatcher($this->server, $this->connections, $this->config);
        $this->databasePersistence = new DatabasePersistence();
        $this->redisPersistence    = new RedisPersistence();
        $this->events();
    }

public function start(): void
    {
        $this->server->start();
    }

protected function events(): void
    {
        $this->open();
        $this->message();
        $this->close();
        $this->server->on('workerStart', function ($server, $workerId) {
            if ($workerId === 0) {
                $this->subscribeToRedis();
            }
        });
    }

protected function open(): void
    {
        $this->server->on('open', function (Server $server, Request $request) {
            $userid = (int)($request->get['userid'] ?? 0);
            if ($userid > 0) {
                if ($this->config['database_persistence']) {
                    $offlineMessages = WebSocketMessage::where('to_userid', $userid)
                        ->where('status', 'UNREAD')
                        ->get();
                    foreach ($offlineMessages as $message) {
                        $message = MessageFormatter::format($message->type, $message->from_userid, $message->to_userid, $message->content);
                        $server->push($request->fd, json_encode($message, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
                        $this->databasePersistence->remove($message->to_userid);
                    }
                }
                if ($this->config['redis_persistence']) {
                    $key = $this->redisPersistence->getUserOfflineMessagesKey($userid);
                    while ($message = \Illuminate\Support\Facades\Redis::lpop($key)) {
                        $server->push($request->fd, $message);
                    }
                }

$this->connections->add($userid, $request->fd);
                $this->heartbeat->pong($request->fd);
            } else {
                $server->close($request->fd);
            }
        });
    }

protected function message(): void
    {
        $this->server->on('message', function (Server $server, $frame) {
            $data = json_decode($frame->data, true);
            if (empty($data)) {
                return;
            }

if ($data['type'] !== 'ping' && empty($data['content'])) {
                return;
            }

if (($data['type'] ?? '') === 'ping') {
                $this->heartbeat->pong($frame->fd);
                $server->push($frame->fd, json_encode(['type' => 'pong']));
                return;
            }

$this->dispatcher->handle($frame->fd, $data);
        });
    }

protected function close(): void
    {
        $this->server->on('close', function (Server $server, $fd) {
            $this->connections->remove($fd);
            $this->heartbeat->remove($fd);
        });
    }

protected function subscribeToRedis(): void
    {
        $dispatcher = $this->dispatcher;
        $config     = $this->config;
        Co::set(['hook_flags' => SWOOLE_HOOK_ALL]);
        Co\run(function () use ($config, $dispatcher) {
            go(function () use ($config, $dispatcher) {
                while (true) {
                    try {
                        $client = new Client([
                            'read_write_timeout' => 0,
                            'protocol' => 3,
                        ]);

$push = $client->push(static function (ClientInterface $client) use ($config) {
                            $client->subscribe($config['redis_subscribe_channel']);
                        });
                        foreach ($push as $notification) {
                            if ((null !== $notification) && $notification->getDataType() === PushResponseInterface::MESSAGE_DATA_TYPE) {
                                $message = $notification[2];
                                $data    = json_decode($message, true);
                                if (empty($data) || empty($data['content'])) {
                                    continue;
                                }
                                $dispatcher->pushSystemMessage($data);
                            }
                        }
                    } catch (\Throwable $throwable) {
                        Log::error('redis-消息订阅异常：', [$throwable->getMessage()]);
                        Co::sleep(3);
                    }
                }
            });
        });
    }
}
```

发布于1年前 · 1 次浏览 · 来自提问

做个俗人

```
<?php

namespace Layman\LaravelWebsocket\Server;

class WebSocketServer
{

public function start(): void
    {
        $this->server->start();
    }

if ($data['type'] !== 'ping' && empty($data['content'])) {
                return;
            }

$this->dispatcher->handle($frame->fd, $data);
        });
    }

登录后参与评论

2025-07-17

Rango-H