本文目录导读：
我来详细介绍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();
}
}
3 测试注意事项
- 浏览器限制：一个域名的并发连接数
- 移动设备电池消耗
- 服务器资源占用
选择SSE还是WebSocket取决于具体需求，SSE更适合单向推送、自动重连的场景，实现简单、支持HTTP/2，适合实时通知、实时数据更新等应用。
