本文目录导读：
我来为您介绍PHP实时数据大屏后端的完整实现方案。
整体架构设计
┌─────────────────────────────────────────────────┐
│ 数据大屏前端（WebSocket客户端） │
└────────────────────┬────────────────────────────┘
│ WebSocket/HTTP
┌────────────────────▼────────────────────────────┐
│ PHP 后端服务（实时推送） │
├────────────────────────────────────────────────┤
│ 1. WebSocket服务 2. RESTful API接口 │
│ 3. 数据聚合服务 4. 缓存服务 │
│ 5. 定时任务服务 6. 消息队列 │
└────────────────────────────────────────────────┘核心功能实现
WebSocket 服务（使用 Workerman）
<?php
use Workerman\Worker;
use Workerman\Timer;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';
class RealtimeDataServer {
private $worker;
private $dataProvider;
private $clients = [];
public function __construct() {
$this->worker = new Worker('websocket://0.0.0.0:2345');
$this->worker->count = 4;
$this->initEventHandlers();
$this->initDataProvider();
}
private function initEventHandlers() {
$this->worker->onConnect = function(TcpConnection $connection) {
echo "Client connected: {$connection->getRemoteIp()}\n";
};
$this->worker->onMessage = function(TcpConnection $connection, $data) {
$message = json_decode($data, true);
switch ($message['type'] ?? '') {
case 'subscribe':
$this->handleSubscribe($connection, $message);
break;
case 'unsubscribe':
$this->handleUnsubscribe($connection, $message);
break;
case 'ping':
$connection->send(json_encode(['type' => 'pong']));
break;
}
};
$this->worker->onClose = function(TcpConnection $connection) {
echo "Client disconnected\n";
$this->removeClient($connection);
};
}
private function initDataProvider() {
// 每5秒推送一次实时数据
Timer::add(5, function() {
$this->broadcastData();
});
// 每分钟推送统计数据
Timer::add(60, function() {
$this->broadcastStatistics();
});
}
private function handleSubscribe($connection, $message) {
$channel = $message['channel'] ?? 'default';
$clientId = spl_object_hash($connection);
$this->clients[$clientId] = [
'connection' => $connection,
'channel' => $channel
];
// 发送初始数据
$initialData = $this->getChannelData($channel);
$connection->send(json_encode([
'type' => 'initial_data',
'channel' => $channel,
'data' => $initialData
]));
}
private function handleUnsubscribe($connection, $message) {
$this->removeClient($connection);
}
private function removeClient($connection) {
$clientId = spl_object_hash($connection);
unset($this->clients[$clientId]);
}
private function broadcastData() {
$data = $this->getRealtimeData();
foreach ($this->clients as $client) {
$client['connection']->send(json_encode([
'type' => 'realtime_data',
'data' => $data,
'timestamp' => time()
]));
}
}
private function broadcastStatistics() {
$stats = $this->getStatistics();
foreach ($this->clients as $client) {
$client['connection']->send(json_encode([
'type' => 'statistics',
'data' => $stats,
'timestamp' => time()
]));
}
}
private function getChannelData($channel) {
// 根据订阅频道返回对应数据
switch ($channel) {
case 'orders':
return $this->getOrdersData();
case 'users':
return $this->getUsersData();
default:
return $this->getRealtimeData();
}
}
private function getRealtimeData() {
// 模拟实时数据，实际项目中从数据库/缓存获取
return [
'online_users' => rand(1000, 5000),
'new_orders' => rand(50, 200),
'sales_amount' => rand(10000, 50000),
'visitors' => rand(500, 2000),
'conversion_rate' => rand(1, 10) / 10,
'regions' => $this->getRegionData()
];
}
private function getStatistics() {
return [
'total_users' => rand(50000, 100000),
'total_orders' => rand(10000, 50000),
'total_sales' => rand(1000000, 5000000),
'avg_order_value' => rand(50, 200),
'hourly_stats' => $this->getHourlyStats()
];
}
private function getRegionData() {
$regions = ['北京', '上海', '广州', '深圳', '杭州', '成都'];
$data = [];
foreach ($regions as $region) {
$data[] = [
'name' => $region,
'value' => rand(100, 1000)
];
}
return $data;
}
private function getHourlyStats() {
$hours = [];
for ($i = 0; $i < 24; $i++) {
$hours[] = [
'hour' => $i,
'orders' => rand(10, 100),
'sales' => rand(100, 1000)
];
}
return $hours;
}
}
// 启动服务
$server = new RealtimeDataServer();
Worker::runAll();
RESTful API 接口
<?php
use Slim\Factory\AppFactory;
use Psr\Http\Message\ResponseInterface as Response;
use Psr\Http\Message\ServerRequestInterface as Request;
require __DIR__ . '/vendor/autoload.php';
class DashboardAPI {
private $app;
private $cache;
private $db;
public function __construct() {
$this->app = AppFactory::create();
$this->initDatabase();
$this->initRoutes();
}
private function initDatabase() {
// 使用PDO连接数据库
$this->db = new PDO(
'mysql:host=localhost;dbname=dashboard;charset=utf8mb4',
'username',
'password'
);
$this->db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
// 初始化Redis缓存
$this->cache = new Redis();
$this->cache->connect('localhost', 6379);
}
private function initRoutes() {
$this->app->get('/api/dashboard/summary', function(Request $request, Response $response) {
$data = $this->getSummaryData();
return $this->jsonResponse($response, $data);
});
$this->app->get('/api/dashboard/realtime', function(Request $request, Response $response) {
$data = $this->getRealtimeData();
return $this->jsonResponse($response, $data);
});
$this->app->get('/api/dashboard/statistics', function(Request $request, Response $response) {
$params = $request->getQueryParams();
$startDate = $params['start_date'] ?? date('Y-m-d', strtotime('-7 days'));
$endDate = $params['end_date'] ?? date('Y-m-d');
$data = $this->getStatistics($startDate, $endDate);
return $this->jsonResponse($response, $data);
});
$this->app->get('/api/dashboard/{metric}', function(Request $request, Response $response, $args) {
$metric = $args['metric'];
$data = $this->getMetricData($metric);
return $this->jsonResponse($response, $data);
});
}
private function getSummaryData() {
// 检查缓存
$cacheKey = 'dashboard:summary';
$cached = $this->cache->get($cacheKey);
if ($cached !== false) {
return json_decode($cached, true);
}
// 从数据库获取数据
$today = date('Y-m-d');
$sql = "SELECT
(SELECT COUNT(*) FROM users WHERE created_at >= ?) as new_users,
(SELECT COUNT(*) FROM orders WHERE created_at >= ?) as today_orders,
(SELECT SUM(total_amount) FROM orders WHERE created_at >= ?) as today_sales,
(SELECT COUNT(*) FROM orders) as total_orders,
(SELECT SUM(total_amount) FROM orders) as total_sales";
$stmt = $this->db->prepare($sql);
$stmt->execute([$today, $today, $today]);
$data = $stmt->fetch(PDO::FETCH_ASSOC);
// 缓存数据（5分钟）
$this->cache->setex($cacheKey, 300, json_encode($data));
return $data;
}
private function getRealtimeData() {
// 实时数据直接从缓存获取（由WebSocket服务更新）
$cacheKey = 'dashboard:realtime';
$cached = $this->cache->get($cacheKey);
if ($cached !== false) {
return json_decode($cached, true);
}
return $this->getFallbackRealtimeData();
}
private function getStatistics($startDate, $endDate) {
$cacheKey = "dashboard:stats:{$startDate}:{$endDate}";
$cached = $this->cache->get($cacheKey);
if ($cached !== false) {
return json_decode($cached, true);
}
// 按天统计数据
$sql = "SELECT
DATE(created_at) as date,
COUNT(*) as orders,
SUM(total_amount) as sales,
COUNT(DISTINCT user_id) as users
FROM orders
WHERE created_at BETWEEN ? AND ?
GROUP BY DATE(created_at)
ORDER BY date";
$stmt = $this->db->prepare($sql);
$stmt->execute([$startDate, $endDate . ' 23:59:59']);
$data = $stmt->fetchAll(PDO::FETCH_ASSOC);
// 缓存数据（1小时）
$this->cache->setex($cacheKey, 3600, json_encode($data));
return $data;
}
private function getMetricData($metric) {
// 根据不同指标返回不同数据
switch ($metric) {
case 'sales':
$data = $this->getSalesMetrics();
break;
case 'users':
$data = $this->getUserMetrics();
break;
case 'system':
$data = $this->getSystemMetrics();
break;
default:
$data = ['status' => 'error', 'message' => 'Unknown metric'];
}
return $data;
}
private function getSalesMetrics() {
return [
'today' => [
'amount' => $this->getTodaySales(),
'orders' => $this->getTodayOrders()
],
'weekly' => $this->getWeeklySales(),
'monthly' => $this->getMonthlySales()
];
}
private function getUserMetrics() {
return [
'active_today' => $this->getActiveUsersToday(),
'active_online' => $this->getOnlineUsers(),
'growth_rate' => $this->getUserGrowthRate()
];
}
private function getSystemMetrics() {
return [
'cpu_usage' => sys_getloadavg()[0],
'memory_usage' => memory_get_usage(true) / 1024 / 1024,
'db_connections' => $this->db->getAttribute(PDO::ATTR_CONNECTION_STATUS)
];
}
private function getTodaySales() {
$stmt = $this->db->query("SELECT SUM(total_amount) FROM orders WHERE DATE(created_at) = CURDATE()");
return (float)$stmt->fetchColumn();
}
private function getTodayOrders() {
$stmt = $this->db->query("SELECT COUNT(*) FROM orders WHERE DATE(created_at) = CURDATE()");
return (int)$stmt->fetchColumn();
}
private function getWeeklySales() {
$stmt = $this->db->query("SELECT DAYOFWEEK(created_at) as day, SUM(total_amount) as sales FROM orders WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY) GROUP BY DAYOFWEEK(created_at)");
return $stmt->fetchAll(PDO::FETCH_ASSOC);
}
private function getMonthlySales() {
$stmt = $this->db->query("SELECT DATE_FORMAT(created_at, '%Y-%m') as month, SUM(total_amount) as sales FROM orders WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 6 MONTH) GROUP BY DATE_FORMAT(created_at, '%Y-%m')");
return $stmt->fetchAll(PDO::FETCH_ASSOC);
}
private function getActiveUsersToday() {
$stmt = $this->db->query("SELECT COUNT(DISTINCT user_id) FROM user_activity WHERE DATE(created_at) = CURDATE()");
return (int)$stmt->fetchColumn();
}
private function getOnlineUsers() {
$cacheKey = 'dashboard:online_users';
$online = $this->cache->get($cacheKey);
return $online !== false ? (int)$online : 0;
}
private function getUserGrowthRate() {
$today = $this->getUserCountToday();
$monthAgo = $this->getUserCountMonthAgo();
if ($monthAgo > 0) {
return (($today - $monthAgo) / $monthAgo) * 100;
}
return 0;
}
private function getUserCountToday() {
$stmt = $this->db->query("SELECT COUNT(*) FROM users WHERE DATE(created_at) = CURDATE()");
return (int)$stmt->fetchColumn();
}
private function getUserCountMonthAgo() {
$stmt = $this->db->query("SELECT COUNT(*) FROM users WHERE DATE(created_at) = DATE_SUB(CURDATE(), INTERVAL 30 DAY)");
return (int)$stmt->fetchColumn();
}
private function getFallbackRealtimeData() {
// 从数据库获取最近的实时数据
$data = [
'online_users' => 0,
'new_orders' => 0,
'sales_amount' => 0,
'visitors' => 0,
'timestamp' => time(),
'source' => 'fallback'
];
try {
$stmt = $this->db->query("
SELECT
(SELECT COUNT(*) FROM active_sessions WHERE online = 1) as online_users,
(SELECT COUNT(*) FROM orders WHERE created_at >= DATE_SUB(NOW(), INTERVAL 5 MINUTE)) as new_orders,
(SELECT SUM(total_amount) FROM orders WHERE created_at >= DATE_SUB(NOW(), INTERVAL 5 MINUTE)) as sales_amount,
(SELECT COUNT(*) FROM visitors WHERE last_seen >= DATE_SUB(NOW(), INTERVAL 5 MINUTE)) as visitors
");
$result = $stmt->fetch(PDO::FETCH_ASSOC);
if ($result) {
$data = array_merge($data, $result);
}
} catch (Exception $e) {
// 记录错误日志
error_log("Error getting fallback data: " . $e->getMessage());
}
return $data;
}
private function jsonResponse(Response $response, $data) {
$response->getBody()->write(json_encode([
'status' => 'success',
'data' => $data
]));
return $response->withHeader('Content-Type', 'application/json');
}
public function run() {
$this->app->run();
}
}
// 启动API服务
$api = new DashboardAPI();
$api->run();
数据提供者（数据源）
<?php
class DataProvider {
private $db;
private $cache;
private $messageQueue;
public function __construct($db, $cache, $messageQueue) {
$this->db = $db;
$this->cache = $cache;
$this->messageQueue = $messageQueue;
}
public function updateRealtimeData() {
// 更新实时数据到缓存
$data = [
'online_users' => $this->getOnlineUsers(),
'new_orders' => $this->getRecentOrders(),
'sales_amount' => $this->getRecentSales(),
'visitors' => $this->getRecentVisitors(),
'timestamp' => time()
];
$this->cache->setex('dashboard:realtime', 5, json_encode($data));
// 发布到消息队列
$this->messageQueue->publish('dashboard', 'realtime', $data);
}
private function getOnlineUsers() {
$stmt = $this->db->query("SELECT COUNT(*) FROM active_sessions WHERE last_activity > DATE_SUB(NOW(), INTERVAL 5 MINUTE)");
return (int)$stmt->fetchColumn();
}
private function getRecentOrders() {
$stmt = $this->db->query("SELECT COUNT(*) FROM orders WHERE created_at > DATE_SUB(NOW(), INTERVAL 5 MINUTE)");
return (int)$stmt->fetchColumn();
}
private function getRecentSales() {
$stmt = $this->db->query("SELECT COALESCE(SUM(total_amount), 0) FROM orders WHERE created_at > DATE_SUB(NOW(), INTERVAL 5 MINUTE)");
return (float)$stmt->fetchColumn();
}
private function getRecentVisitors() {
$stmt = $this->db->query("SELECT COUNT(*) FROM visitors WHERE last_seen > DATE_SUB(NOW(), INTERVAL 5 MINUTE)");
return (int)$stmt->fetchColumn();
}
public function startUpdater() {
// 使用定时任务更新数据
while (true) {
$this->updateRealtimeData();
sleep(5); // 每5秒更新一次
}
}
}
消息队列配置（使用Redis）
<?php
use Predis\Client;
class MessageQueue {
private $redis;
private $subscriptions;
public function __construct(Client $redis) {
$this->redis = $redis;
$this->subscriptions = [];
}
public function publish($channel, $event, $data) {
$message = json_encode([
'event' => $event,
'data' => $data,
'timestamp' => time()
]);
$this->redis->publish($channel, $message);
}
public function subscribe($channel, callable $callback) {
$this->subscriptions[$channel][] = $callback;
}
public function listen() {
$loop = new \React\EventLoop\StreamSelectLoop();
$client = new \React\Socket\SocketClient('127.0.0.1:6379');
$this->redis->psubscribe(['dashboard_*'], function($redis, $pattern, $channel, $message) {
foreach ($this->subscriptions[$channel] ?? [] as $callback) {
$callback(json_decode($message, true));
}
});
}
}
数据库表结构设计
-- 用户表
CREATE TABLE users (
id INT PRIMARY KEY AUTO_INCREMENT,
username VARCHAR(50) NOT NULL UNIQUE,
email VARCHAR(100) NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_login TIMESTAMP NULL
);
-- 订单表
CREATE TABLE orders (
id INT PRIMARY KEY AUTO_INCREMENT,
user_id INT NOT NULL,
total_amount DECIMAL(10,2) NOT NULL,
status VARCHAR(20) DEFAULT 'pending',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
paid_at TIMESTAMP NULL,
FOREIGN KEY (user_id) REFERENCES users(id)
);
-- 访问者表
CREATE TABLE visitors (
id INT PRIMARY KEY AUTO_INCREMENT,
session_id VARCHAR(64) NOT NULL UNIQUE,
ip_address VARCHAR(45),
page_url VARCHAR(255),
visited_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_seen TIMESTAMP NULL
);
-- 实时统计数据表
CREATE TABLE realtime_stats (
id INT PRIMARY KEY AUTO_INCREMENT,
metric_name VARCHAR(50) NOT NULL,
metric_value DECIMAL(15,2) NOT NULL,
recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_metric_name_time (metric_name, recorded_at)
);
-- 用户活跃记录表
CREATE TABLE user_activity (
id INT PRIMARY KEY AUTO_INCREMENT,
user_id INT NOT NULL,
action VARCHAR(50) NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id)
);
部署与配置
启动脚本
<?php // start_server.php - 主服务启动脚本 require_once __DIR__ . '/vendor/autoload.php'; use Workerman\Worker; // 启动WebSocket服务 $realtimeServer = new RealtimeDataServer(); Worker::runAll(); // 启动RESTful API $api = new DashboardAPI();
Nginx 配置
server {
listen 80;
server_name dashboard.example.com;
# RESTful API
location /api {
proxy_pass http://127.0.0.1:8080;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header Host $http_host;
}
# WebSocket
location /ws {
proxy_pass http://127.0.0.1:2345;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $http_host;
proxy_set_header X-Real-IP $remote_addr;
}
# 静态文件
location / {
root /var/www/dashboard/frontend;
index index.html;
}
}
Docker 部署配置
# Dockerfile
FROM php:8.0-apache
RUN apt-get update && apt-get install -y \
libpq-dev \
redis-server \
supervisor \
&& docker-php-ext-install pdo pdo_mysql \
&& docker-php-ext-enable pdo pdo_mysql
WORKDIR /var/www/html
COPY . /var/www/html
RUN curl -sS https://getcomposer.org/installer | php \
&& mv composer.phar /usr/local/bin/composer \
&& composer install --no-dev
# 安装 Workerman
RUN composer require workerman/workerman \
&& composer require slim/slim:^4.0
# 启动脚本
COPY docker-entrypoint.sh /docker-entrypoint.sh
RUN chmod +x /docker-entrypoint.sh
EXPOSE 80 2345 6379
CMD ["/docker-entrypoint.sh"]
# docker-compose.yml
version: '3.8'
services:
db:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: rootpassword
MYSQL_DATABASE: dashboard
MYSQL_USER: dashboard
MYSQL_PASSWORD: dashboard123
volumes:
- db_data:/var/lib/mysql
ports:
- "3306:3306"
redis:
image: redis:alpine
ports:
- "6379:6379"
app:
build: .
ports:
- "80:80"
- "2345:2345"
environment:
DB_HOST: db
DB_NAME: dashboard
DB_USER: dashboard
DB_PASS: dashboard123
REDIS_HOST: redis
CACHE_TTL: 300
depends_on:
- db
- redis
volumes:
- ./:/var/www/html
restart: unless-stopped
workers:
build: .
command: php websocket_server.php
environment:
DB_HOST: db
REDIS_HOST: redis
depends_on:
- app
- db
- redis
volumes:
- ./:/var/www/html
restart: unless-stopped
volumes:
db_data:
性能优化建议
- 使用Redis/APCu缓存减少数据库查询
- 数据库索引优化 - 定期运行EXPLAIN分析查询
- 启用OPcache提高PHP执行效率
- 使用CDN缓存静态资源
- 数据库读写分离 - 主库负责写入，从库负责读取
- 使用连接池管理数据库连接
- 消息队列解耦数据产生和处理逻辑
监控与日志
<?php
class DashboardLogger {
private $logDir;
private $file;
public function __construct($logDir = '/var/log/dashboard') {
$this->logDir = $logDir;
if (!is_dir($logDir)) {
mkdir($logDir, 0777, true);
}
$this->file = $logDir . '/dashboard_' . date('Y-m-d') . '.log';
}
public function log($message, $level = 'INFO') {
$logEntry = sprintf(
"[%s] [%s] %s\n",
date('Y-m-d H:i:s'),
$level,
$message
);
file_put_contents($this->file, $logEntry, FILE_APPEND);
}
public function error($message) {
$this->log($message, 'ERROR');
}
public function warning($message) {
$this->log($message, 'WARNING');
}
public function info($message) {
$this->log($message, 'INFO');
}
}
这个完整的PHP实时数据大屏后端方案提供了：
- 实时数据推送（WebSocket）
- RESTful API
- 数据缓存
- 消息队列
- 数据库设计
- 部署配置
- 日志监控
您可以根据实际需求调整数据源、推送频率和接口设计。
