Symfony6中PHP通过WebSocket向客户端发消息的正确流程及问题排查

你的问题根源在于控制器进程与WebSocket服务器进程完全隔离，直接在控制器中注入WebSocketServer实例调用广播方法无效，同时用sleep模拟异步会阻塞HTTP请求，导致WebSocket消息无法及时推送到客户端。以下是具体修复方案：

1. 重构异步任务处理 #

放弃控制器内的sleep阻塞，改用Symfony Messenger组件将后台任务丢入队列，控制器立即返回响应，避免阻塞WebSocket连接。

步骤1：创建后台任务消息类 #

// src/Message/ProcessJarTask.php
namespace App\Message;

class ProcessJarTask
{
}

步骤2：修改控制器，发送任务到队列 #

/**
 * @Route("/start-process", name="start_process", methods={"POST"})
 */
public function startProcess(Request $request, MessageBusInterface $messageBus): JsonResponse
{
    // 将Jar文件处理任务投递到后台队列
    $messageBus->dispatch(new ProcessJarTask());

    // 立即返回响应，不阻塞客户端
    return $this->json(['message' => '后台任务已启动']);
}

步骤3：创建任务处理器，处理Jar文件并触发WebSocket消息 #

// src/MessageHandler/ProcessJarTaskHandler.php
namespace App\MessageHandler;

use App\Message\ProcessJarTask;
use App\Message\WebSocketBroadcastMessage;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Messenger\MessageBusInterface;

#[AsMessageHandler]
class ProcessJarTaskHandler
{
    private $messageBus;

    public function __construct(MessageBusInterface $messageBus)
    {
        $this->messageBus = $messageBus;
    }

    public function __invoke(ProcessJarTask $task)
    {
        // 模拟Jar文件执行过程
        $this->messageBus->dispatch(new WebSocketBroadcastMessage('progress', '任务启动...'));
        sleep(2);
        
        $this->messageBus->dispatch(new WebSocketBroadcastMessage('progress', '任务执行中...'));
        sleep(2);
        
        $this->messageBus->dispatch(new WebSocketBroadcastMessage('progress', '任务完成'));
    }
}

2. 让WebSocket服务器监听消息队列 #

WebSocket服务器需要独立运行，并监听Messenger的广播消息，再推送给客户端。以下是基于Ratchet的示例实现：

步骤1：创建WebSocket广播消息类 #

// src/Message/WebSocketBroadcastMessage.php
namespace App\Message;

class WebSocketBroadcastMessage
{
    private $type;
    private $text;

    public function __construct(string $type, string $text)
    {
        $this->type = $type;
        $this->text = $text;
    }

    public function getType(): string
    {
        return $this->type;
    }

    public function getText(): string
    {
        return $this->text;
    }
}

步骤2：编写WebSocket服务器命令 #

// src/Command/WebSocketServerCommand.php
namespace App\Command;

use Ratchet\ConnectionInterface;
use Ratchet\MessageComponentInterface;
use Ratchet\Server\IoServer;
use Ratchet\Http\HttpServer;
use Ratchet\WebSocket\WsServer;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Messenger\MessageBusInterface;
use App\Message\WebSocketBroadcastMessage;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

class WebSocketServerCommand extends Command implements MessageComponentInterface
{
    protected static $defaultName = 'app:websocket:start';
    private $connections = [];
    private $messageBus;

    public function __construct(MessageBusInterface $messageBus)
    {
        parent::__construct();
        $this->messageBus = $messageBus;
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $server = IoServer::factory(
            new HttpServer(new WsServer($this)),
            8080
        );

        $output->writeln('WebSocket服务器已启动: ws://localhost:8080');
        $server->run();

        return Command::SUCCESS;
    }

    public function onOpen(ConnectionInterface $conn)
    {
        $this->connections[$conn->resourceId] = $conn;
    }

    public function onMessage(ConnectionInterface $from, $msg) {}

    public function onClose(ConnectionInterface $conn)
    {
        unset($this->connections[$conn->resourceId]);
    }

    public function onError(ConnectionInterface $conn, \Exception $e)
    {
        $conn->close();
    }

    #[AsMessageHandler]
    public function handleBroadcast(WebSocketBroadcastMessage $message)
    {
        $payload = json_encode([
            'type' => $message->getType(),
            'text' => $message->getText()
        ]);

        // 广播消息给所有连接的客户端
        foreach ($this->connections as $conn) {
            $conn->send($payload);
        }
    }
}

3. 启动服务与验证 #

1. 启动WebSocket服务器：

php bin/console app:websocket:start

2. 启动Messenger消费进程（处理后台任务）：

php bin/console messenger:consume async

3. 运行Symfony应用，点击按钮后，客户端控制台将依次收到任务进度消息。

内容的提问来源于stack exchange，提问作者jose gilberto loucel molina