PHPAPISSE推送怎么实现

wen PHP项目 2026-07-06 193

本文目录导读：

PHPAPISSE推送怎么实现

基本工作原理
服务器端PHP实现
客户端JavaScript实现
高级特性实现
优化和注意事项
完整生产环境示例
调试技巧

在PHP中实现SSE（Server-Sent Events）推送，核心是利用PHP的流式输出特性，让客户端通过EventSource API持续接收服务器推送的数据,以下是详细的实现步骤和示例代码：

基本工作原理

SSE是单向通信协议，服务器持续发送数据，客户端通过EventSource接收，PHP实现的关键是：设置正确的响应头、保持连接、循环发送数据。

服务器端PHP实现

基础SSE推送代码 (sse.php)

<?php
// 禁用输出缓冲
ob_end_clean();
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');
header('X-Accel-Buffering: no'); // Nginx专用，禁用缓冲
// 允许跨域（如果需要）
header('Access-Control-Allow-Origin: *');
// 设置执行时间限制为无限
set_time_limit(0);
// 循环发送事件
while (true) {
    // 获取当前时间
    $data = [
        'time' => date('Y-m-d H:i:s'),
        'message' => 'Hello at ' . date('H:i:s')
    ];
    // 发送事件（格式：event: xxx\ndata: xxx\n\n）
    echo "event: timeUpdate\n";
    echo "data: " . json_encode($data) . "\n\n";
    // 立即输出
    ob_flush();
    flush();
    // 等待1秒
    sleep(1);
    // 检测客户端是否断开连接
    if (connection_aborted()) {
        break;
    }
}
?>

带事件类型的完整示例 (sse_advanced.php)

<?php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');
header('X-Accel-Buffering: no');
set_time_limit(0);
$lastId = 0;
while (true) {
    // 检查客户端连接状态
    if (connection_aborted()) {
        break;
    }
    $lastId++;
    // 随机生成数据
    $type = rand(1, 3);
    $data = [];
    switch ($type) {
        case 1:
            $data = ['type' => 'notification', 'message' => '新消息', 'id' => $lastId];
            echo "event: notification\n";
            break;
        case 2:
            $data = ['type' => 'update', 'count' => rand(10, 100), 'id' => $lastId];
            echo "event: update\n";
            break;
        case 3:
            $data = ['type' => 'heartbeat', 'timestamp' => time(), 'id' => $lastId];
            echo "event: heartbeat\n";
            break;
    }
    // 发送数据
    echo "id: {$lastId}\n";
    echo "data: " . json_encode($data) . "\n\n";
    ob_flush();
    flush();
    sleep(2);
}
?>

客户端JavaScript实现

基本EventSource使用 (index.html)

<!DOCTYPE html>
<html>
<head>SSE Demo</title>
</head>
<body>
    <h1>SSE实时推送</h1>
    <div id="messages"></div>
    <script>
        // 创建EventSource连接
        const eventSource = new EventSource('sse.php');
        // 监听默认事件（没有指定event名称的）
        eventSource.onmessage = function(event) {
            const data = JSON.parse(event.data);
            console.log('收到数据:', data);
            displayMessage(data);
        };
        // 监听自定义事件
        eventSource.addEventListener('timeUpdate', function(event) {
            const data = JSON.parse(event.data);
            console.log('时间更新:', data);
        });
        eventSource.addEventListener('notification', function(event) {
            const data = JSON.parse(event.data);
            console.log('通知:', data);
            displayMessage(data);
        });
        eventSource.addEventListener('update', function(event) {
            const data = JSON.parse(event.data);
            console.log('更新:', data);
        });
        eventSource.addEventListener('heartbeat', function(event) {
            const data = JSON.parse(event.data);
            console.log('心跳:', data);
        });
        // 连接打开
        eventSource.onopen = function(event) {
            console.log('SSE连接已建立');
        };
        // 错误处理
        eventSource.onerror = function(event) {
            console.error('SSE连接错误:', event);
            if (eventSource.readyState === EventSource.CLOSED) {
                console.log('连接已关闭，尝试重新连接...');
                // 可以在这里实现重新连接逻辑
                setTimeout(() => {
                    location.reload();
                }, 3000);
            }
        };
        function displayMessage(data) {
            const div = document.getElementById('messages');
            const p = document.createElement('p');
            p.textContent = JSON.stringify(data);
            div.appendChild(p);
        }
    </script>
</body>
</html>

高级特性实现

1 带断线重连和最后ID (sse_reconnect.php)

<?php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');
set_time_limit(0);
// 获取客户端最后接收的ID
$lastEventId = isset($_SERVER['HTTP_LAST_EVENT_ID']) ? 
                intval($_SERVER['HTTP_LAST_EVENT_ID']) : 0;
$currentId = $lastEventId;
// 模拟数据源
$dataSource = [
    ['id' => 1, 'message' => '第一条消息'],
    ['id' => 2, 'message' => '第二条消息'],
    ['id' => 3, 'message' => '第三条消息'],
];
while (true) {
    if (connection_aborted()) break;
    $currentId++;
    // 模拟根据ID获取新数据
    $newData = getDataSince($currentId);
    if ($newData) {
        echo "id: {$currentId}\n";
        echo "event: newData\n";
        echo "data: " . json_encode($newData) . "\n\n";
        ob_flush();
        flush();
    }
    // 重连时间（客户端等待2秒后重连）
    echo "retry: 2000\n\n";
    ob_flush();
    flush();
    sleep(3);
}
function getDataSince($lastId) {
    global $dataSource;
    foreach ($dataSource as &$data) {
        if ($data['id'] > $lastId) {
            return $data;
        }
    }
    return null;
}
?>

2 集成到Laravel框架 (SSEController.php)

<?php
namespace App\Http\Controllers;
use Illuminate\Http\Request;
use Symfony\Component\HttpFoundation\StreamedResponse;
class SSEController extends Controller
{
    public function stream()
    {
        $response = new StreamedResponse(function () {
            // 禁用输出缓冲
            ob_end_clean();
            while (true) {
                if (connection_aborted()) {
                    break;
                }
                $data = [
                    'timestamp' => now()->toDateTimeString(),
                    'users_online' => rand(50, 200),
                    'new_orders' => rand(0, 10)
                ];
                echo "event: dashboardUpdate\n";
                echo "data: " . json_encode($data) . "\n\n";
                ob_flush();
                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');
        return $response;
    }
}

优化和注意事项

1 Nginx配置优化

# 在server或location块中添加
proxy_buffering off;
fastcgi_buffering off;  # PHP-FPM使用

2 Apache配置

# 在.htaccess或虚拟主机配置中
SetEnv no-gzip 1

3 性能优化建议

使用消息队列：对于复杂场景，使用Redis/Beanstalkd等消息队列
多进程处理：PHP单进程处理大量连接时，考虑使用Swoole/Workerman
连接池管理：维护连接池管理大量SSE连接
心跳机制：定期发送心跳包保持连接

4 安全考虑

// 身份验证示例
$token = $_GET['token'] ?? '';
if (!validateToken($token)) {
    header('HTTP/1.1 401 Unauthorized');
    echo "data: {error: 'Unauthorized'}\n\n";
    exit;
}

完整生产环境示例

使用Redis消息队列 (sse_redis.php)

<?php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');
set_time_limit(0);
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
// 订阅频道
$channel = 'notifications';
while (true) {
    if (connection_aborted()) break;
    // 阻塞等待消息
    $message = $redis->brpop($channel, 2); // 2秒超时
    if ($message) {
        $data = json_decode($message[1], true);
        echo "id: {$data['id']}\n";
        echo "event: {$data['type']}\n";
        echo "data: " . json_encode($data['payload']) . "\n\n";
        ob_flush();
        flush();
    } else {
        // 发送心跳
        echo ": heartbeat\n\n";
        ob_flush();
        flush();
    }
}
?>

调试技巧

客户端调试（Chrome开发者工具）

Network标签查看请求
EventStream标签查看事件流
Console查看EventSource事件

服务端调试

// 在开发环境添加日志
$log = fopen('sse_debug.log', 'a');
fwrite($log, date('Y-m-d H:i:s') . " - 发送数据\n");

SSE推送在PHP中的实现虽然简单，但生产环境中需要考虑性能、扩展性和稳定性，对于高并发场景，建议使用Swoole或Workerman等异步框架，或者采用消息队列+多进程的方案,保持客户端断线重连机制和服务器心跳检测是保证服务稳定性的关键。

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