PHP 实时数据大屏后端

本文目录导读：

1. 整体架构设计
2. 核心功能实现
3. 部署与配置

我来为您介绍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
• 数据缓存
• 消息队列
• 数据库设计
• 部署配置
• 日志监控

您可以根据实际需求调整数据源、推送频率和接口设计。