本文目录导读：
我来详细讲解如何在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);
}
}
}
最佳实践
- 使用HTTPS：生产环境必须使用HTTPS
- 合理设置超时：配置适当的SSE超时时间
- 错误处理：实现重连机制和错误处理
- 负载均衡：使用Redis或其他共享状态支持多实例
- 监控：监控Mercure Hub的健康状态
- 主题命名规范：使用标准的URL格式命名Topic
- 数据压缩：对于大量数据考虑压缩传输
这样配置完成后,你的Symfony项目就具备了完整的Mercure实时推送能力。
