PHP 实时数据接口

wen PHP项目 2026-08-10 5

本文目录导读：

PHP 实时数据接口

SSE (Server-Sent Events) 实时推送
WebSocket 实时通信
轮询实时接口
通用RESTful API接口
最佳实践建议

我将为您提供一个完整的PHP实时数据接口解决方案,包含多种实现方式。

SSE (Server-Sent Events) 实时推送

服务端实现 (sse_server.php)

<?php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');
header('X-Accel-Buffering: no'); // 禁用Nginx缓冲
// 关闭执行时间限制
set_time_limit(0);
// 确保数据实时输出
while (ob_get_level() > 0) {
    ob_end_flush();
}
ob_implicit_flush(true);
// 连接断开检测
ignore_user_abort(false);
echo "retry: 3000\n"; // 重连间隔3秒
// 模拟实时数据推送
$counter = 0;
while (!connection_aborted()) {
    $counter++;
    // 获取实时数据（这里可以是数据库查询、API调用等）
    $data = [
        'id' => $counter,
        'timestamp' => date('Y-m-d H:i:s'),
        'value' => rand(1, 100),
        'message' => "第{$counter}条实时数据"
    ];
    // 发送数据
    echo "id: {$counter}\n";
    echo "data: " . json_encode($data, JSON_UNESCAPED_UNICODE) . "\n\n";
    // 刷新输出缓冲
    flush();
    // 每2秒推送一次
    sleep(2);
    // 测试用：到达10条后停止
    if ($counter >= 10) {
        echo "event: close\ndata: {\"message\":\"连接关闭\"}\n\n";
        flush();
        break;
    }
}
?>

客户端实现 (sse_client.php)

<!DOCTYPE html>
<html>
<head>SSE 实时数据</title>
    <style>
        #data-container {
            max-height: 400px;
            overflow-y: scroll;
            border: 1px solid #ddd;
            padding: 10px;
            margin: 20px;
        }
        .data-item {
            margin: 10px 0;
            padding: 10px;
            background: #f5f5f5;
            border-radius: 5px;
        }
    </style>
</head>
<body>
    <h2>SSE 实时数据流</h2>
    <div id="data-container"></div>
    <script>
        // 连接状态显示
        const container = document.getElementById('data-container');
        // 创建EventSource连接
        const eventSource = new EventSource('sse_server.php');
        // 连接打开
        eventSource.onopen = function(e) {
            container.innerHTML += '<div class="data-item" style="color:green;">✓ 连接成功</div>';
        };
        // 接收普通消息
        eventSource.onmessage = function(e) {
            const data = JSON.parse(e.data);
            console.log('收到数据:', data);
            const div = document.createElement('div');
            div.className = 'data-item';
            div.innerHTML = `
                <strong>ID:</strong> ${data.id}<br>
                <strong>时间:</strong> ${data.timestamp}<br>
                <strong>值:</strong> ${data.value}<br>
                <strong>消息:</strong> ${data.message}
            `;
            container.appendChild(div);
            // 自动滚动到底部
            container.scrollTop = container.scrollHeight;
        };
        // 自定义事件
        eventSource.addEventListener('close', function(e) {
            container.innerHTML += '<div class="data-item" style="color:red;">× 连接关闭</div>';
            eventSource.close();
        });
        // 错误处理
        eventSource.onerror = function(e) {
            console.error('连接错误:', e);
            container.innerHTML += '<div class="data-item" style="color:red;">⚠ 连接断开，正在重连...</div>';
        };
        // 手动断开连接
        function disconnect() {
            eventSource.close();
            container.innerHTML += '<div class="data-item" style="color:orange;">✕ 已手动断开连接</div>';
        }
    </script>
</body>
</html>

WebSocket 实时通信

WebSocket 服务器 (websocket_server.php)

<?php
// 简单的WebSocket服务器实现
class WebSocketServer {
    private $clients = [];
    private $socket;
    public function __construct($host = '0.0.0.0', $port = 8080) {
        $this->socket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
        socket_set_option($this->socket, SOL_SOCKET, SO_REUSEADDR, 1);
        socket_bind($this->socket, $host, $port);
        socket_listen($this->socket);
        echo "WebSocket服务器启动在 ws://$host:$port\n";
    }
    public function run() {
        while (true) {
            $read = array_merge([$this->socket], $this->clients);
            $write = $except = null;
            socket_select($read, $write, $except, null);
            foreach ($read as $socket) {
                if ($socket === $this->socket) {
                    // 新连接
                    $client = socket_accept($this->socket);
                    if ($client) {
                        $this->clients[] = $client;
                        echo "新客户端连接，当前客户端数: " . count($this->clients) . "\n";
                    }
                } else {
                    // 处理客户端数据
                    $data = @socket_read($socket, 1024, PHP_BINARY_READ);
                    if ($data === false) {
                        // 连接关闭
                        $this->removeClient($socket);
                    } else if ($data) {
                        // 处理WebSocket握手
                        if (strpos($data, 'Sec-WebSocket-Key') !== false) {
                            $this->handshake($socket, $data);
                        } else {
                            // 处理消息
                            $decoded = $this->decodeFrame($data);
                            echo "收到消息: " . $decoded . "\n";
                            // 发送实时数据给所有客户端
                            $response = [
                                'type' => 'realtime',
                                'data' => $this->getRealtimeData(),
                                'timestamp' => time()
                            ];
                            $this->sendToAll(json_encode($response));
                        }
                    }
                }
            }
        }
    }
    private function handshake($client, $request) {
        preg_match('/Sec-WebSocket-Key:\s(.*?)\r\n/', $request, $matches);
        $key = $matches[1];
        $accept = base64_encode(sha1($key . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', true));
        $response = "HTTP/1.1 101 Switching Protocols\r\n";
        $response .= "Upgrade: websocket\r\n";
        $response .= "Connection: Upgrade\r\n";
        $response .= "Sec-WebSocket-Accept: {$accept}\r\n\r\n";
        socket_write($client, $response);
    }
    private function decodeFrame($data) {
        $secondByte = ord($data[1]);
        $length = $secondByte & 127;
        if ($length <= 125) {
            $maskStart = 2;
        } else if ($length == 126) {
            $length = unpack('n', substr($data, 2, 2))[1];
            $maskStart = 4;
        } else {
            $length = unpack('N', substr($data, 2, 4))[1];
            $maskStart = 10;
        }
        $masks = substr($data, $maskStart, 4);
        $data = substr($data, $maskStart + 4);
        $decoded = '';
        for ($i = 0; $i < strlen($data); $i++) {
            $decoded .= $data[$i] ^ $masks[$i % 4];
        }
        return $decoded;
    }
    private function encodeFrame($message) {
        $frame = [];
        $len = strlen($message);
        $frame[0] = 0x81; // text frame
        if ($len <= 125) {
            $frame[1] = $len;
        } else if ($len <= 65535) {
            $frame[1] = 126;
            $frame[2] = ($len >> 8) & 255;
            $frame[3] = $len & 255;
        } else {
            $frame[1] = 127;
            for ($i = 7; $i >= 0; $i--) {
                $frame[$i + 2] = ($len >> ($i * 8)) & 255;
            }
        }
        $frame = array_merge($frame, unpack('C*', $message));
        return call_user_func_array('pack', array_merge(['C*'], $frame));
    }
    private function getRealtimeData() {
        // 这里可以替换为真实的数据库查询或API调用
        return [
            'id' => time(),
            'value' => rand(1, 1000),
            'data' => date('Y-m-d H:i:s')
        ];
    }
    private function sendToAll($message) {
        foreach ($this->clients as $client) {
            socket_write($client, $this->encodeFrame($message));
        }
    }
    private function removeClient($socket) {
        $key = array_search($socket, $this->clients);
        if ($key !== false) {
            unset($this->clients[$key]);
            socket_close($socket);
            echo "客户端断开，剩余客户端数: " . count($this->clients) . "\n";
        }
    }
    public function __destruct() {
        socket_close($this->socket);
    }
}
// 启动服务器
$server = new WebSocketServer('localhost', 8080);
$server->run();
?>

WebSocket 客户端 (websocket_client.php)

<!DOCTYPE html>
<html>
<head>WebSocket 实时数据</title>
    <style>
        #logs {
            margin: 20px;
            padding: 10px;
            border: 1px solid #ddd;
            max-height: 400px;
            overflow-y: auto;
        }
        .log-item {
            padding: 5px;
            margin: 5px 0;
            background: #f9f9f9;
        }
    </style>
</head>
<body>
    <h2>WebSocket 实时数据</h2>
    <button onclick="connect()">连接</button>
    <button onclick="disconnect()">断开</button>
    <button onclick="sendData()">发送数据</button>
    <div id="logs"></div>
    <script>
        let ws = null;
        const logs = document.getElementById('logs');
        function addLog(message) {
            const div = document.createElement('div');
            div.className = 'log-item';
            div.textContent = message;
            logs.appendChild(div);
            logs.scrollTop = logs.scrollHeight;
        }
        function connect() {
            if (ws && ws.readyState === WebSocket.OPEN) {
                addLog('已连接');
                return;
            }
            ws = new WebSocket('ws://localhost:8080');
            ws.onopen = function() {
                addLog('✓ 已连接到WebSocket服务器');
            };
            ws.onmessage = function(event) {
                addLog('收到数据: ' + event.data);
            };
            ws.onerror = function(error) {
                addLog('❌ 发生错误: ' + error.message);
            };
            ws.onclose = function() {
                addLog('连接已关闭');
            };
        }
        function disconnect() {
            if (ws) {
                ws.close();
                addLog('已断开连接');
            }
        }
        function sendData() {
            if (ws && ws.readyState === WebSocket.OPEN) {
                const data = {
                    type: 'request',
                    data: '获取实时数据'
                };
                ws.send(JSON.stringify(data));
                addLog('发送数据: ' + JSON.stringify(data));
            } else {
                addLog('未连接服务器');
            }
        }
    </script>
</body>
</html>

轮询实时接口

轮询接口 (polling_api.php)

<?php
// API响应
header('Content-Type: application/json');
// 获取请求参数
$lastId = isset($_GET['last_id']) ? (int)$_GET['last_id'] : 0;
// 模拟数据库查询（这里使用文件存储来模拟）
$dataFile = 'data.json';
// 生成或获取数据
$data = [
    'last_id' => $lastId + 1,
    'items' => [],
    'timestamp' => date('Y-m-d H:i:s')
];
// 模拟生成新数据
for ($i = 1; $i <= 3; $i++) {
    $data['items'][] = [
        'id' => $lastId + $i,
        'value' => rand(1000, 9999),
        'time' => date('Y-m-d H:i:s')
    ];
}
// 返回JSON数据
echo json_encode($data);
?>

轮询客户端

<!DOCTYPE html>
<html>
<head>轮询实时数据</title>
    <script>
        let lastId = 0;
        // 定期轮询
        function pollData() {
            fetch(`polling_api.php?last_id=${lastId}`)
                .then(response => response.json())
                .then(data => {
                    if (data.items.length > 0) {
                        lastId = data.last_id;
                        console.log('添加新数据:', data.items);
                    }
                })
                .catch(error => console.error('轮询错误:', error));
        }
        // 每3秒轮询一次
        setInterval(pollData, 3000);
        // 立即执行一次
        pollData();
    </script>
</head>
<body>
    <h2>轮询实时数据</h2>
</body>
</html>

通用RESTful API接口

REST API (api.php)

<?php
header('Content-Type: application/json');
header('Access-Control-Allow-Origin: *');
header('Access-Control-Allow-Methods: GET, POST, PUT, DELETE, OPTIONS');
header('Access-Control-Allow-Headers: Content-Type, Authorization');
// 简单的路由处理
$method = $_SERVER['REQUEST_METHOD'];
$path = isset($_GET['path']) ? $_GET['path'] : '/';
// 数据库连接（示例使用PDO）
$pdo = new PDO('mysql:host=localhost;dbname=test', 'username', 'password');
switch ($method) {
    case 'GET':
        if ($path === 'realtime-data') {
            // 获取实时数据
            $stmt = $pdo->prepare("SELECT * FROM realtime_data ORDER BY id DESC LIMIT 10");
            $stmt->execute();
            $data = $stmt->fetchAll(PDO::FETCH_ASSOC);
            echo json_encode([
                'success' => true,
                'data' => $data,
                'timestamp' => time()
            ]);
        }
        break;
    case 'POST':
        // 添加新数据
        $input = json_decode(file_get_contents('php://input'), true);
        $stmt = $pdo->prepare("INSERT INTO realtime_data (value, description) VALUES (?, ?)");
        $stmt->execute([$input['value'], $input['description']]);
        echo json_encode([
            'success' => true,
            'message' => '数据添加成功',
            'id' => $pdo->lastInsertId()
        ]);
        break;
    case 'PUT':
        // 更新数据
        break;
    case 'DELETE':
        // 删除数据
        break;
    default:
        http_response_code(405);
        echo json_encode(['error' => 'Method not allowed']);
}
?>

最佳实践建议

配置与优化 (config.php)

<?php
// 实时数据配置
return [
    'sse' => [
        'interval' => 2,           // 推送间隔（秒）
        'retry_time' => 3000,       // 重连时间（毫秒）
        'max_duration' => 3600      // 最大持续时间
    ],
    'websocket' => [
        'host' => '0.0.0.0',
        'port' => 8080,
        'ping_interval' => 30,      // 心跳间隔
        'max_connections' => 1000
    ],
    'polling' => [
        'default_interval' => 5,   // 轮询间隔（秒）
        'max_interval' => 60,      // 最大间隔
        'http_client' => [
            'timeout' => 10,
            'connect_timeout' => 5
        ]
    ],
    'cache' => [
        'enabled' => true,
        'driver' => 'redis',
        'ttl' => 30               // 缓存生命周期（秒）
    ],
    'security' => [
        'api_key' => 'your-secret-key',
        'rate_limit' => 100,      // 每分钟请求限制
        'ip_whitelist' => []      // IP白名单
    ]
];
?>

这些实现方案各有优缺点：

SSE：简单，自动化重连，适合服务器到客户端的单向推送
WebSocket：双向通信，实时性最好，适合聊天、游戏等应用
轮询：兼容性最好，实现简单，适合低频更新
REST API：通用性强，适合标准CRUD操作

选择哪种方案取决于您的具体需求、服务器架构和客户端环境，建议根据实际场景选择合适的方案，或者组合使用。