Symfony7中RabbitMQ消息处理器如何在Twig模板显示Flash消息？

首先明确：可行，但不能直接用你当前的实现方式。原因是RabbitMQ的消息处理器是异步后台执行的，和用户发起的HTTP请求不属于同一个上下文——Flash消息依赖用户Session，仅存在于当前请求周期内，异步处理器无法直接操作用户的Session或向当前页面推送消息。

下面是两种可行的实现方案：

方案一：状态存储 + 前端轮询（简单易实现） #

核心思路是：用户发起文件上传请求时生成唯一任务ID，异步处理器将处理状态存储到Redis/数据库，前端定时查询状态并更新页面。

1. 上传控制器（生成任务ID并初始化状态） #

#[Route('/upload', name: 'upload')]
public function upload(Request $request, MessageBusInterface $bus, \Redis $redis): Response
{
    // 处理文件上传逻辑（省略）
    $taskId = Uuid::v4()->toString();
    $sessionId = $request->getSession()->getId();

    // 初始化处理状态到Redis（1小时过期）
    $redis->setex(
        "task:$taskId",
        3600,
        json_encode([
            'status' => 'processing',
            'message' => '文件处理中，请稍候...',
            'session_id' => $sessionId
        ])
    );

    // 发送消息时携带任务ID
    $bus->dispatch(new CSVFileMessage($filePath, $taskId));

    // 把任务ID存入Session，供前端查询使用
    $request->getSession()->set('current_task_id', $taskId);

    return $this->render('upload/status.html.twig');
}

2. 修改消息处理器（更新处理状态） #

#[AsMessageHandler]
final class StatusMessageHandler
{
    private $redis;

    public function __construct(\Redis $redis)
    {
        $this->redis = $redis;
    }

    public function __invoke(CSVFileMessage $message)
    {
        try {
            // 执行文件处理逻辑（省略）

            // 更新状态为成功
            $this->redis->setex(
                "task:{$message->getTaskId()}",
                3600,
                json_encode([
                    'status' => 'success',
                    'message' => '文件处理完成！'
                ])
            );
        } catch (\Exception $e) {
            // 更新状态为失败
            $this->redis->setex(
                "task:{$message->getTaskId()}",
                3600,
                json_encode([
                    'status' => 'error',
                    'message' => "文件处理失败：{$e->getMessage()}"
                ])
            );
        }
    }
}

3. 新增状态查询接口 #

#[Route('/task/status', name: 'task_status')]
public function getTaskStatus(Request $request, \Redis $redis): JsonResponse
{
    $taskId = $request->getSession()->get('current_task_id');
    if (!$taskId) {
        return new JsonResponse(['status' => 'no_task']);
    }

    $taskData = $redis->get("task:$taskId");
    if (!$taskData) {
        return new JsonResponse(['status' => 'not_found']);
    }

    return new JsonResponse(json_decode($taskData, true));
}

4. Twig模板（结合JS轮询更新） #

{% extends 'base.html.twig' %}

{% block body %}
    <div id="status-alert" class="alert alert-info alert-dismissible fade show" role="alert">
        文件处理中，请稍候...
        <button type="button" class="btn-close" data-bs-dismiss="alert" aria-label="Close"></button>
    </div>

    <script>
        // 每2秒查询一次状态
        const statusInterval = setInterval(() => {
            fetch('{{ path('task_status') }}')
                .then(res => res.json())
                .then(data => {
                    const alert = document.getElementById('status-alert');
                    
                    if (data.status === 'success') {
                        alert.className = 'alert alert-success alert-dismissible fade show';
                        alert.textContent = data.message;
                        clearInterval(statusInterval); // 停止轮询
                    } else if (data.status === 'error') {
                        alert.className = 'alert alert-danger alert-dismissible fade show';
                        alert.textContent = data.message;
                        clearInterval(statusInterval);
                    }
                });
        }, 2000);
    </script>
{% endblock %}

方案二：WebSocket实时推送（更实时） #

如果需要更即时的状态更新，可以用Symfony WebSocket组件（如Ratchet），直接从异步处理器推送消息到用户页面：

1. 安装WebSocket相关依赖，配置服务器。
2. 用户打开页面时建立WebSocket连接，关联Session ID。
3. 消息处理器处理完成后，通过WebSocket服务器向对应Session的用户推送状态消息。
4. 前端监听WebSocket消息，动态更新页面内容。

这种方式配置复杂度稍高，但体验更流畅，适合对实时性要求高的场景。

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