Symfony中Ratchet/Pawl WebSocket客户端发送消息失败求助

• Pawl的WebSocket连接是绑定到特定进程/实例的：你在命令行中启动的连接仅存在于该命令进程内，Symfony的其他服务（如控制器、业务服务）运行在独立的请求进程或服务实例中，无法直接复用这个已建立的连接。
• 重建连接会导致原监听失效：每次新建的WebSocket连接都是独立实例，原命令行中绑定的监听逻辑仅属于旧连接，新连接不会继承该逻辑，自然无法继续接收消息。

不需要立刻切换到ZeroMQ，优先尝试以下方案：

1. 将WebSocket连接封装为全局单例服务 #

在Symfony中配置持久化单例服务，确保命令行和其他服务复用同一个连接实例：

• 服务配置（config/services.yaml）：

services:
    App\Service\PersistentWebSocketClient:
        shared: true
        arguments:
            $wsServerUrl: '%env(WEBSOCKET_SERVER_URL)%'

• 服务类实现：

namespace App\Service;

use Ratchet\Client\WebSocket;
use React\EventLoop\LoopInterface;
use Ratchet\Client\Connector;

class PersistentWebSocketClient
{
    private ?WebSocket $connection = null;
    private LoopInterface $loop;
    private Connector $connector;
    private string $wsUrl;

    public function __construct(string $wsUrl, LoopInterface $loop)
    {
        $this->wsUrl = $wsUrl;
        $this->loop = $loop;
        $this->connector = new Connector($this->loop);
    }

    // 命令行中调用此方法初始化连接
    public function startListening(): void
    {
        if ($this->connection) return;

        $this->connector($this->wsUrl)->then(function (WebSocket $conn) {
            $this->connection = $conn;
            
            // 绑定消息监听逻辑
            $conn->on('message', function ($msg) {
                echo "收到服务器消息: {$msg}\n";
            });

            // 连接断开时重置实例，方便后续重连
            $conn->on('close', function ($code, $reason) {
                $this->connection = null;
                echo "连接已关闭: {$code} - {$reason}\n";
            });
        }, function ($e) {
            echo "连接失败: {$e->getMessage()}\n";
        });

        $this->loop->run();
    }

    // 其他服务调用此方法发送消息
    public function send(string $message): bool
    {
        if (!$this->connection) {
            $this->reconnect();
            if (!$this->connection) return false;
        }

        $this->connection->send($message);
        return true;
    }

    // 自动重连逻辑
    private function reconnect(): void
    {
        $this->connection?->close();
        $this->startListening();
    }
}

• 命令行启动类：

namespace App\Command;

use App\Service\PersistentWebSocketClient;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;

class WebSocketListenCommand extends Command
{
    protected static $defaultName = 'app:websocket:listen';

    public function __construct(private PersistentWebSocketClient $client)
    {
        parent::__construct();
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $output->writeln('WebSocket监听已启动...');
        $this->client->startListening();
        return Command::SUCCESS;
    }
}

这样其他服务注入PersistentWebSocketClient后调用send()，就能复用命令行中已建立的连接。

2. 消息队列中转（多进程场景） #

如果你的Symfony应用是多进程部署（如FPM多进程），单例服务无法跨进程共享连接，可使用消息队列（如Redis、RabbitMQ）：

• 命令行进程维持WebSocket连接，同时监听消息队列
• 其他服务需要发送消息时，将消息推送到队列
• 命令行进程从队列取出消息，通过已有的WebSocket连接发送

Symfony可通过symfony/messenger组件快速实现此方案，复杂度低于ZeroMQ。

3. ZeroMQ的适用场景 #

ZeroMQ适合进程间通信（IPC）或跨进程消息传递，如果你需要在Symfony多进程与WebSocket监听进程间直接通信，它是可行方案，但配置复杂度高于上述两种方案。优先用单例或消息队列解决问题，再考虑ZeroMQ。

• Pawl依赖ReactPHP事件循环，必须在长驻进程（如命令行）中运行循环，不能在请求生命周期（如控制器）中启动循环，否则会阻塞请求
• 务必实现连接断开后的自动重连逻辑，确保send()时连接可用

内容的提问来源于stack exchange，提问作者prit