本文目录导读：
在 Laravel 中实现 SSE (Server-Sent Events) 响应，需要合理设置响应头并保持连接持久化,以下是完整的实现方案：
路由配置
// routes/web.php 或 routes/api.php
Route::get('/sse/events', [SSEController::class, 'stream']);
控制器实现
<?php
namespace App\Http\Controllers;
use Illuminate\Http\Request;
use Symfony\Component\HttpFoundation\StreamedResponse;
class SSEController extends Controller
{
public function stream(Request $request)
{
// 设置 SSE 响应头
$response = new StreamedResponse(function () {
// 禁用输出缓冲
if (ob_get_level()) {
ob_end_clean();
}
// 设置无限执行时间
set_time_limit(0);
// 发送初始数据
$this->sendSSEData('connected', [
'message' => '连接成功',
'timestamp' => now()->toIso8601String()
]);
// 循环发送数据
while (true) {
// 检查客户端是否断开连接
if (connection_aborted()) {
break;
}
// 发送事件数据
$this->sendSSEData('message', [
'time' => now()->toIso8601String(),
'data' => 'Some real-time data'
]);
// 发送注释保持连接
echo ": keepalive\n\n";
flush();
// 等待一段时间
sleep(2);
}
});
// 设置响应头
$response->headers->set('Content-Type', 'text/event-stream');
$response->headers->set('Cache-Control', 'no-cache');
$response->headers->set('Connection', 'keep-alive');
$response->headers->set('X-Accel-Buffering', 'no'); // Nginx 特定
return $response;
}
private function sendSSEData($event, $data)
{
echo "event: {$event}\n";
echo "data: " . json_encode($data) . "\n\n";
flush();
}
}
完整的高级实现
<?php
namespace App\Http\Controllers;
use Illuminate\Http\Request;
use Symfony\Component\HttpFoundation\StreamedResponse;
class AdvancedSSEController extends Controller
{
public function stream(Request $request)
{
return response()->stream(function () use ($request) {
// 初始化
$this->initializeStream();
$lastEventId = $request->header('Last-Event-ID', 0);
$userId = $request->get('user_id');
// 发送重连ID
echo "retry: 3000\n\n";
flush();
while (true) {
if (connection_aborted()) break;
// 获取新数据
$events = $this->getNewEvents($lastEventId, $userId);
foreach ($events as $event) {
// 发送事件ID用于断线重连
echo "id: {$event['id']}\n";
echo "event: {$event['type']}\n";
echo "data: " . json_encode($event['data']) . "\n\n";
flush();
$lastEventId = $event['id'];
}
// 定时心跳
echo ": heartbeat\n\n";
flush();
// 控制频率
sleep(3);
}
}, 200, [
'Content-Type' => 'text/event-stream',
'Cache-Control' => 'no-cache',
'Connection' => 'keep-alive',
'X-Accel-Buffering' => 'no',
]);
}
private function initializeStream()
{
// 关闭输出缓冲
while (ob_get_level() > 0) {
ob_end_flush();
}
set_time_limit(0);
ignore_user_abort(false);
}
private function getNewEvents($lastEventId, $userId)
{
// 从数据库获取新事件
// 示例实现
$events = [];
$newEvents = cache()->get("sse_events_{$userId}", []);
foreach ($newEvents as $event) {
if ($event['id'] > $lastEventId) {
$events[] = $event;
}
}
return $events;
}
}
前端 JavaScript 实现
class SSEConnection {
constructor(url) {
this.url = url;
this.eventSource = null;
this.reconnectAttempts = 0;
this.maxReconnectAttempts = 5;
this.lastEventId = 0;
}
connect() {
this.eventSource = new EventSource(this.url);
// 连接打开
this.eventSource.onopen = (event) => {
console.log('SSE 连接已建立');
this.reconnectAttempts = 0;
};
// 接收消息
this.eventSource.onmessage = (event) => {
const data = JSON.parse(event.data);
this.handleMessage(data);
this.lastEventId = event.lastEventId || 0;
};
// 自定义事件
this.eventSource.addEventListener('connected', (event) => {
console.log('连接成功:', JSON.parse(event.data));
});
this.eventSource.addEventListener('message', (event) => {
const data = JSON.parse(event.data);
this.updateUI(data);
});
// 错误处理
this.eventSource.onerror = (error) => {
console.error('SSE 错误:', error);
this.handleReconnection();
};
}
handleReconnection() {
this.eventSource.close();
if (this.reconnectAttempts < this.maxReconnectAttempts) {
this.reconnectAttempts++;
const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
console.log(`尝试重连 (${this.reconnectAttempts}/${this.maxReconnectAttempts})...`);
setTimeout(() => {
this.connect();
}, delay);
} else {
console.log('达到最大重连次数');
this.emitError('连接失败');
}
}
disconnect() {
if (this.eventSource) {
this.eventSource.close();
this.eventSource = null;
}
}
}
// 使用示例
const sse = new SSEConnection('/sse/events?user_id=123');
sse.connect();
// 页面卸载时断开连接
window.addEventListener('beforeunload', () => {
sse.disconnect();
});
Nginx 配置优化
# 确保 Nginx 不缓冲 SSE 响应
location /sse/ {
proxy_pass http://your-app;
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 86400s;
proxy_send_timeout 86400s;
# 禁用缓冲
fastcgi_buffering off;
proxy_set_header Connection '';
chunked_transfer_encoding on;
# 保持连接
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
}
注意事项
-
性能优化：
- 使用队列系统处理事件（Redis、RabbitMQ）
- 避免在循环中查询数据库
- 考虑使用
yield和生成器
-
错误处理：
try { while (true) { // 发送数据 } } catch (\Exception $e) { Log::error('SSE Error: ' . $e->getMessage()); } finally { flush(); } -
资源清理：
register_shutdown_function(function () { // 清理资源 cache()->forget('sse_connection_' . $userId); });
这样就可以在 Laravel 中正确实现 SSE 响应了，如果需要更复杂的事件处理，建议结合 Laravel 的事件系统和队列来管理。
