PHP项目Symfony Mercure实时推送

wen PHP项目 2026-07-24 98

本文目录导读：

Mercure简介
安装和配置
发布消息
前端订阅实现
高级用法
Docker配置
安全注意事项
最佳实践

我来详细讲解如何在Symfony项目中使用Mercure实现实时推送功能。

Mercure简介

Mercure是一个实时推送协议,基于Server-Sent Events (SSE)构建，用于实现服务器到客户端的实时数据推送。

安装和配置

安装Mercure组件

composer require symfony/mercure-bundle

配置Mercure

config/packages/mercure.yaml

mercure:
    hubs:
        default:
            url: '%env(MERCURE_URL)%'  # Mercure Hub地址
            public_url: '%env(MERCURE_PUBLIC_URL)%'  # 公开URL
            jwt:
                secret: '%env(MERCURE_JWT_SECRET)%'  # JWT密钥
                publish: '*'
                subscribe: '*'

环境变量配置

.env文件

# Docker环境
MERCURE_URL=http://mercure:3000/.well-known/mercure
MERCURE_PUBLIC_URL=https://example.com/.well-known/mercure
MERCURE_JWT_SECRET=your-secret-key
# 或使用Symfony Docker
MERCURE_URL=http://localhost:3000/.well-known/mercure
MERCURE_PUBLIC_URL=http://localhost:3000/.well-known/mercure

发布消息

在控制器中发布消息

// src/Controller/NotificationController.php
namespace App\Controller;
use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
use Symfony\Component\Mercure\HubInterface;
use Symfony\Component\Mercure\Update;
use Symfony\Component\HttpFoundation\Response;
use Symfony\Component\Routing\Annotation\Route;
class NotificationController extends AbstractController
{
    #[Route('/send-notification', name: 'send_notification')]
    public function sendNotification(HubInterface $hub): Response
    {
        // 创建更新消息
        $update = new Update(
            'https://example.com/notifications',  // Topic
            json_encode([                          // Data
                'message' => '新的通知！',
                'type' => 'info',
                'timestamp' => time()
            ])
        );
        // 发布到Hub
        $hub->publish($update);
        return $this->json(['status' => 'success']);
    }
}

带Cookie认证的发布

// 使用Cookie进行认证
use Symfony\Component\Mercure\Jwt\TokenFactoryInterface;
class AdminController extends AbstractController
{
    #[Route('/admin/broadcast', name: 'admin_broadcast')]
    public function broadcast(
        HubInterface $hub, 
        TokenFactoryInterface $tokenFactory
    ): Response {
        // 生成JWT令牌
        $token = $tokenFactory->create([
            'mercure' => [
                'publish' => ['*']
            ]
        ]);
        // 设置Cookie
        $response = new Response();
        $response->headers->setCookie(
            new Cookie(
                'mercureAuthorization',
                $token,
                time() + 3600,
                '/.well-known/mercure',
                null,
                false,
                true,
                false,
                'lax'
            )
        );
        return $response;
    }
}

前端订阅实现

使用EventSource API

// public/js/mercure-subscriber.js
class MercureSubscriber {
    constructor(hubUrl, topics) {
        this.hubUrl = hubUrl;
        this.topics = topics;
        this.eventSource = null;
    }
    connect() {
        // 构建URL
        const url = new URL(this.hubUrl);
        this.topics.forEach(topic => {
            url.searchParams.append('topic', topic);
        });
        // 建立连接
        this.eventSource = new EventSource(url.toString());
        // 处理消息
        this.eventSource.onmessage = (event) => {
            const data = JSON.parse(event.data);
            console.log('收到消息:', data);
            this.handleMessage(data);
        };
        // 处理连接打开
        this.eventSource.onopen = () => {
            console.log('Mercure连接已建立');
        };
        // 处理错误
        this.eventSource.onerror = (error) => {
            console.error('Mercure连接错误:', error);
            // 自动重连
            setTimeout(() => this.connect(), 5000);
        };
    }
    handleMessage(data) {
        // 在这里处理接收到的消息
        switch(data.type) {
            case 'info':
                this.showNotification(data.message);
                break;
            case 'warning':
                this.showWarning(data.message);
                break;
            case 'error':
                this.showError(data.message);
                break;
            default:
                console.log('未知消息类型:', data);
        }
    }
    showNotification(message) {
        // 显示通知
        const notification = document.createElement('div');
        notification.className = 'alert alert-info';
        notification.textContent = message;
        document.getElementById('notifications').appendChild(notification);
        // 自动移除
        setTimeout(() => notification.remove(), 5000);
    }
    disconnect() {
        if (this.eventSource) {
            this.eventSource.close();
        }
    }
}
// 使用示例
const subscriber = new MercureSubscriber(
    'http://localhost:3000/.well-known/mercure',
    ['https://example.com/notifications']
);
subscriber.connect();

Fetch API方式（现代浏览器）

// 使用Fetch API的流式响应
async function subscribeToMercure(hubUrl, topics) {
    const url = new URL(hubUrl);
    topics.forEach(topic => {
        url.searchParams.append('topic', topic);
    });
    try {
        const response = await fetch(url.toString(), {
            headers: {
                'Accept': 'text/event-stream',
                'Cache-Control': 'no-cache'
            }
        });
        const reader = response.body.getReader();
        const decoder = new TextDecoder();
        while (true) {
            const { done, value } = await reader.read();
            if (done) break;
            const chunk = decoder.decode(value);
            // 解析SSE格式
            const events = parseSSEChunk(chunk);
            events.forEach(event => {
                if (event.data) {
                    handleEvent(JSON.parse(event.data));
                }
            });
        }
    } catch (error) {
        console.error('连接失败:', error);
        // 自动重连
        setTimeout(() => subscribeToMercure(hubUrl, topics), 5000);
    }
}
function parseSSEChunk(chunk) {
    const events = [];
    const lines = chunk.split('\n');
    let currentEvent = {};
    for (const line of lines) {
        if (line.startsWith('data: ')) {
            currentEvent.data = line.slice(6);
        } else if (line.startsWith('event: ')) {
            currentEvent.event = line.slice(7);
        } else if (line === '') {
            if (Object.keys(currentEvent).length > 0) {
                events.push(currentEvent);
                currentEvent = {};
            }
        }
    }
    return events;
}

高级用法

带认证的私有Topic

// 创建私有Topic的更新
class PrivateNotificationController extends AbstractController
{
    #[Route('/private-notification/{userId}', name: 'private_notification')]
    public function sendPrivateNotification(
        int $userId, 
        HubInterface $hub
    ): Response {
        // 为用户创建私有Topic
        $topic = sprintf('https://example.com/user/%d/notifications', $userId);
        $update = new Update(
            $topic,
            json_encode([
                'message' => '这是私有消息',
                'userId' => $userId
            ]),
            true  // private标志
        );
        $hub->publish($update);
        return $this->json(['status' => 'sent']);
    }
}

批量发布消息

use Symfony\Component\Mercure\Update;
use Symfony\Component\Mercure\HubInterface;
class BatchNotificationController extends AbstractController
{
    #[Route('/batch-notification', name: 'batch_notification')]
    public function batchNotification(HubInterface $hub): Response
    {
        $updates = [];
        // 创建多个更新
        for ($i = 0; $i < 10; $i++) {
            $updates[] = new Update(
                'https://example.com/notifications',
                json_encode([
                    'id' => $i,
                    'message' => "消息 #{$i}",
                    'timestamp' => time()
                ])
            );
        }
        // 批量发布
        foreach ($updates as $update) {
            $hub->publish($update);
        }
        return $this->json(['status' => 'batch sent', 'count' => count($updates)]);
    }
}

Symfony Messenger集成

// src/Message/NotificationMessage.php
class NotificationMessage
{
    public function __construct(
        private string $topic,
        private array $data
    ) {}
    public function getTopic(): string
    {
        return $this->topic;
    }
    public function getData(): array
    {
        return $this->data;
    }
}
// src/MessageHandler/NotificationHandler.php
class NotificationHandler implements MessageHandlerInterface
{
    public function __construct(
        private HubInterface $hub
    ) {}
    public function __invoke(NotificationMessage $message)
    {
        $update = new Update(
            $message->getTopic(),
            json_encode($message->getData())
        );
        $this->hub->publish($update);
    }
}
// 在控制器中发送
class SomeController extends AbstractController
{
    public function someAction(MessageBusInterface $bus): Response
    {
        $bus->dispatch(new NotificationMessage(
            'https://example.com/notifications',
            ['message' => '异步通知']
        ));
        return $this->json(['status' => 'queued']);
    }
}

Docker配置

# docker-compose.yml
version: '3.8'
services:
  mercure:
    image: dunglas/mercure
    restart: unless-stopped
    ports:
      - "3000:3000"
    environment:
      SERVER_NAME: ':3000'
      MERCURE_PUBLISHER_JWT_KEY: '!ChangeThisMercureHubJWTSecretKey!'
      MERCURE_SUBSCRIBER_JWT_KEY: '!ChangeThisMercureHubJWTSecretKey!'
      MERCURE_EXTRA_DIRECTIVES: |-
        cors_origins http://localhost:8080
        anonymous
    volumes:
      - mercure_data:/data
    networks:
      - app-network
volumes:
  mercure_data:
networks:
  app-network:
    driver: bridge

安全注意事项

JWT安全配置

# config/packages/mercure.yaml
mercure:
    hubs:
        default:
            url: '%env(MERCURE_URL)%'
            jwt:
                secret: '%env(MERCURE_JWT_SECRET)%'
                publish:
                    - 'https://example.com/notifications'
                    - 'https://example.com/users/*'
                subscribe:
                    - 'https://example.com/users/{subscriber}'

验证订阅者身份

// 验证订阅者身份
class MercureSubscriber implements EventSubscriberInterface
{
    public static function getSubscribedEvents()
    {
        return [
            MercureEvents::PUBLISH => 'onPublish',
        ];
    }
    public function onPublish(MercureEvent $event)
    {
        $update = $event->getUpdate();
        // 检查发布者权限
        if (!$this->security->isGranted('ROLE_ADMIN')) {
            $event->setPropagationStopped(true);
        }
    }
}