PHP SSE服务端推送

wen PHP项目 2026-08-11 60

本文目录导读：

PHP SSE服务端推送

SSE基础概念
基础实现
高级应用模式
性能优化与最佳实践
完整示例：通知系统
注意事项

我来详细介绍PHP实现SSE（Server-Sent Events）服务端推送的技术方案。

SSE基础概念

什么是SSE？

服务器向客户端单向推送数据的web技术
基于HTTP长连接
自动重连机制
比WebSocket更轻量，实现更简单

基础实现

1 简单的SSE服务端

<?php
// sse.php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');
header('X-Accel-Buffering: no'); // 禁用Nginx缓冲
// 禁用PHP执行时间限制
set_time_limit(0);
// 关闭输出缓冲
while (ob_get_level()) {
    ob_end_flush();
}
ob_implicit_flush(true);
function sendEvent($data, $event = 'message', $id = null) {
    if ($id !== null) {
        echo "id: {$id}\n";
    }
    if ($event !== 'message') {
        echo "event: {$event}\n";
    }
    echo "data: " . json_encode($data) . "\n\n";
    flush();
}
// 循环推送数据
$counter = 0;
while (true) {
    $counter++;
    // 发送消息
    sendEvent([
        'time' => date('H:i:s'),
        'counter' => $counter,
        'message' => '这是第' . $counter . '条消息'
    ], 'message', $counter);
    // 每2秒发送一次
    sleep(2);
    // 可以设置结束条件
    if ($counter >= 10) break;
}
?>

2 完整的前端客户端

<!DOCTYPE html>
<html>
<head>SSE Demo</title>
</head>
<body>
    <h1>SSE 实时数据</h1>
    <div id="output"></div>
    <script>
        // 创建EventSource连接
        const eventSource = new EventSource('sse.php');
        // 处理连接打开
        eventSource.onopen = function(e) {
            console.log('连接已建立');
            addMessage('系统', '连接成功');
        };
        // 处理默认消息
        eventSource.onmessage = function(e) {
            const data = JSON.parse(e.data);
            addMessage('消息', data.message, e.lastEventId);
        };
        // 处理自定义事件
        eventSource.addEventListener('custom', function(e) {
            const data = JSON.parse(e.data);
            addMessage('自定义', data.message);
        });
        // 处理错误
        eventSource.onerror = function(e) {
            console.log('连接错误，准备重连...', e);
            addMessage('错误', '连接中断，正在重连...');
        };
        function addMessage(type, message, id) {
            const div = document.getElementById('output');
            const p = document.createElement('p');
            p.innerHTML = `<strong>${type}:</strong> ${message} ${id ? '(ID: ' + id + ')' : ''}`;
            div.appendChild(p);
        }
        // 关闭连接
        function closeConnection() {
            eventSource.close();
            addMessage('系统', '连接已关闭');
        }
    </script>
</body>
</html>

高级应用模式

1 事件驱动异步推送

<?php
// async_sse.php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');
// 使用Redis订阅模式实现事件驱动
class SSEEventManager {
    private $redis;
    public function __construct() {
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
    }
    public function run() {
        // 记录客户端连接
        $clientId = $_GET['client'] ?? uniqid();
        // 订阅频道
        $this->redis->subscribe(['channel_' . $clientId, 'global'], function($redis, $channel, $message) {
            if ($message === 'PING') return;
            echo "data: {$message}\n\n";
            flush();
        });
    }
    // 推送消息到特定客户端
    public function pushToClient($clientId, $data) {
        $this->redis->publish('channel_' . $clientId, json_encode($data));
    }
}
?>

2 支持多用户分区推送

<?php
// multi_user_sse.php
class MultiUserSSE {
    private $sessionId;
    private $lastEventId;
    public function __construct() {
        $this->sessionId = $_GET['session'] ?? 'default';
        $this->lastEventId = $_GET['lastEventId'] ?? null;
    }
    public function stream() {
        $this->setupHeaders();
        while (true) {
            // 从数据库或缓存获取新数据
            $events = $this->fetchNewEvents();
            if (!empty($events)) {
                foreach ($events as $event) {
                    $this->sendEvent($event);
                }
            }
            // 发送心跳包
            echo ": heartbeat\n\n";
            flush();
            // 检查客户端是否断开
            if (connection_aborted()) {
                break;
            }
            sleep(5);
        }
    }
    private function fetchNewEvents() {
        // 从Redis或数据库获取增量事件
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        $events = [];
        $key = "events:{$this->sessionId}";
        // 获取新事件
        $lastId = $this->lastEventId ?? 0;
        $newEvents = $redis->zRangeByScore($key, $lastId + 1, '+inf');
        foreach ($newEvents as $event) {
            $events[] = json_decode($event, true);
        }
        return $events;
    }
    private function sendEvent($event) {
        $id = $event['id'];
        $eventType = $event['type'] ?? 'message';
        echo "id: {$id}\n";
        echo "event: {$eventType}\n";
        echo "data: " . json_encode($event['data']) . "\n\n";
        flush();
    }
    private function setupHeaders() {
        header('Content-Type: text/event-stream');
        header('Cache-Control: no-cache');
        header('Connection: keep-alive');
        header('X-Accel-Buffering: no');
        set_time_limit(0);
        while (ob_get_level()) {
            ob_end_flush();
        }
        ob_implicit_flush(true);
    }
}
?>

3 与数据库事件集成

<?php
// db_sse.php
class DatabaseSSE {
    private $db;
    public function __construct() {
        $this->db = new PDO('mysql:host=localhost;dbname=test', 'root', '');
    }
    public function streamNewRecords() {
        header('Content-Type: text/event-stream');
        header('Cache-Control: no-cache');
        $lastId = 0;
        while (true) {
            try {
                // 查询新增记录
                $stmt = $this->db->prepare("
                    SELECT *, id FROM messages 
                    WHERE id > ? AND status = 'new'
                    ORDER BY id ASC
                    LIMIT 10
                ");
                $stmt->execute([$lastId]);
                $newRecords = $stmt->fetchAll(PDO::FETCH_ASSOC);
                foreach ($newRecords as $record) {
                    $lastId = $record['id'];
                    echo "id: {$record['id']}\n";
                    echo "event: new-record\n";
                    echo "data: " . json_encode($record) . "\n\n";
                    flush();
                    // 标记已处理
                    $updateStmt = $this->db->prepare("
                        UPDATE messages SET status = 'sent' WHERE id = ?
                    ");
                    $updateStmt->execute([$record['id']]);
                }
                echo ": keepalive\n\n";
                flush();
            } catch (Exception $e) {
                echo "event: error\n";
                echo "data: " . json_encode(['message' => $e->getMessage()]) . "\n\n";
                flush();
            }
            if (connection_aborted()) {
                break;
            }
            sleep(2);
        }
    }
}
// 后台推送脚本示例
// push_message.php
if ($_SERVER['REQUEST_METHOD'] === 'POST') {
    $message = $_POST['message'] ?? '';
    // 确保所有已连接的客户端收到消息
    $redis = new Redis();
    $redis->connect('127.0.0.1', 6379);
    $redis->publish('global', json_encode([
        'time' => time(),
        'message' => $message
    ]));
    echo 'Message pushed';
}
?>

性能优化与最佳实践

1 使用长连接池

<?php
// optimized_sse.php
class OptimizedSSE {
    private $keepAliveTime = 30;
    public function start() {
        $this->setupEnvironment();
        // 发送初始消息确认连接
        $this->send('connected', ['status' => 'ok', 'time' => time()]);
        $lastKeepAlive = time();
        while (true) {
            // 获取新数据（这里使用Redis）
            $data = $this->getIncrementalData();
            if ($data) {
                $this->send('update', $data);
            }
            // 心跳包维护连接
            $now = time();
            if ($now - $lastKeepAlive >= $this->keepAliveTime) {
                echo ": heartbeat\n\n";
                flush();
                $lastKeepAlive = $now;
            }
            // 检测客户端断开
            if (connection_aborted()) {
                $this->cleanup();
                break;
            }
            usleep(100000); // 100ms
        }
    }
    private function getIncrementalData() {
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        // 从Redis列表获取新数据
        $data = $redis->lPop('events_queue');
        return $data ? json_decode($data, true) : null;
    }
}
?>

2 Nginx配置优化

# nginx.conf 配置
location /sse {
    proxy_pass http://php-backend;
    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 3600s;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding on;
    # 禁用压缩
    gzip off;
    # 允许CORS
    add_header Access-Control-Allow-Origin *;
    add_header X-Accel-Buffering no;
}

完整示例：通知系统

<?php
// notification_system.php
class NotificationSystem {
    private $userId;
    private $redis;
    private $db;
    public function __construct($userId) {
        $this->userId = $userId;
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
    }
    public function stream() {
        // 设置SSE头
        header('Content-Type: text/event-stream');
        header('Cache-Control: no-cache');
        header('X-Accel-Buffering: no');
        // 订阅用户通知频道
        $channel = "user:{$this->userId}:notif";
        $this->redis->subscribe([$channel], function($redis, $channel, $message) {
            $notification = json_decode($message, true);
            echo "id: {$notification['id']}\n";
            echo "event: notification\n";
            echo "data: " . json_encode([
                'type' => $notification['type'],
                'content' => $notification['content'],
                'time' => $notification['created_at']
            ]) . "\n\n";
            flush();
        });
    }
    public function sendNotification($content, $type = 'info') {
        $notification = [
            'id' => uniqid(),
            'user_id' => $this->userId,
            'content' => $content,
            'type' => $type,
            'created_at' => date('Y-m-d H:i:s')
        ];
        // 存储通知到数据库
        $this->saveToDatabase($notification);
        // 发布到Redis频道
        $json = json_encode($notification);
        $this->redis->publish("user:{$this->userId}:notif", $json);
    }
    private function saveToDatabase($notification) {
        // 数据库操作...
    }
}
?>

注意事项

1 内存和资源管理

设置合理的执行时间限制
定期清理资源和关闭数据库连接
处理客户端断线情况

2 错误处理

try {
    // SSE逻辑
} catch (Exception $e) {
    echo "event: error\n";
    echo "data: " . json_encode(['message' => '服务器错误']) . "\n\n";
    flush();
} finally {
    // 清理资源
    if (isset($redis)) {
        $redis->close();
    }
}