LaravelSSE响应怎么设置

wen PHP项目 2026-07-06 179

本文目录导读：

LaravelSSE响应怎么设置

路由配置
控制器实现
完整的高级实现
前端 JavaScript 实现
Nginx 配置优化
注意事项

在 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 的事件系统和队列来管理。

本文地址： https://www.bmjup.com/post/17354.html