PHP项目队列监控如何采集生产消费指标数据

wen PHP项目 2026-06-25 28

本文目录导读：

核心指标定义
数据采集实现方案
实时监控实现
日志采集方案
可视化展示
最佳实践建议
实际部署架构

针对 PHP 项目的队列监控，采集生产消费指标数据通常涉及以下几个核心层面，以下是具体的实现方案和工具推荐：

核心指标定义

首先明确需要采集的关键指标：

指标类型	指标名称	说明
生产端	生产速率	每秒/分钟入队数量
生产端	生产延迟	消息从创建到入队的时间
队列	队列深度	当前待处理消息数
队列	延迟队列大小	延迟消息数量
消费端	消费速率	每秒/分钟处理数量
消费端	消费延迟	消息入队到被消费的时间
消费端	失败率	处理失败占比
消费端	处理耗时	平均/最大处理时间
系统	Worker数量	当前活跃消费者数

数据采集实现方案

使用 Redis + Laravel Horizon（推荐）

如果你的项目基于 Laravel，Horizon 提供了完善的监控能力：

// config/horizon.php
'metrics' => [
    'snapshots' => [
        // 采集频率配置
        'rate' => 5, // 每5秒采集一次
    ],
],
// 自定义采集逻辑
class QueueMetricsCollector
{
    public function collect(): array
    {
        return [
            'queue_size' => $this->getQueueSize(),
            'processing' => $this->getProcessingCount(),
            'failed' => $this->getFailedCount(),
            'throughput' => $this->getThroughput(),
        ];
    }
    private function getThroughput(): array
    {
        $redis = Redis::connection('horizon');
        $recent = $redis->zcount('horizon:jobs:recent', '-inf', '+inf');
        $timeWindow = 60; // 60秒窗口
        return [
            'rate' => $recent / $timeWindow,
            'total' => $recent,
        ];
    }
}

通用 PHP 实现（基于 Redis）

class QueueMonitor
{
    private Redis $redis;
    private string $prefix = 'queue:metrics:';
    public function recordProduction(string $queue, array $data): void
    {
        $now = microtime(true);
        $minute = floor($now / 60) * 60;
        // 速率统计（每分钟）
        $this->redis->hIncrBy(
            "{$this->prefix}prod_rate:{$queue}:{$minute}",
            'count',
            1
        );
        // 记录生产延迟
        $this->redis->lPush(
            "{$this->prefix}prod_latency:{$queue}",
            $now - $data['created_at'] // 消息创建到入队的时间
        );
        // 设置过期时间（保留24小时）
        $this->redis->expire("{$this->prefix}prod_rate:{$queue}:{$minute}", 86400);
        $this->redis->lTrim("{$this->prefix}prod_latency:{$queue}", 0, 999);
        $this->redis->expire("{$this->prefix}prod_latency:{$queue}", 3600);
    }
    public function recordConsumption(string $queue, array $data): void
    {
        $now = microtime(true);
        $minute = floor($now / 60) * 60;
        $cost = ($now - $data['pushed_at']) * 1000; // 处理耗时(ms)
        // 消费速率
        $this->redis->hIncrBy(
            "{$this->prefix}consume_rate:{$queue}:{$minute}",
            'count',
            1
        );
        // 处理耗时统计（P50, P95, P99）
        $this->redis->lPush(
            "{$this->prefix}cost:{$queue}",
            json_encode([
                'cost' => $cost,
                'time' => $now
            ])
        );
        // 消费延迟
        $this->redis->lPush(
            "{$this->prefix}consume_latency:{$queue}",
            $now - $data['pushed_at']
        );
        // 成功/失败统计
        $status = $data['success'] ? 'success' : 'failed';
        $this->redis->hIncrBy(
            "{$this->prefix}status:{$queue}:{$minute}",
            $status,
            1
        );
    }
    public function getMetrics(string $queue): array
    {
        $now = time();
        $metrics = [];
        // 当前队列深度
        $metrics['queue_depth'] = $this->redis->lLen("queue:{$queue}");
        // 最近5分钟生产速率
        for ($i = 0; $i < 5; $i++) {
            $minute = floor(($now - $i * 60) / 60) * 60;
            $count = $this->redis->hGet(
                "{$this->prefix}prod_rate:{$queue}:{$minute}",
                'count'
            ) ?: 0;
            $metrics['prod_rate'][] = [
                'time' => $minute,
                'count' => (int)$count
            ];
        }
        // 平均处理耗时
        $costs = $this->redis->lRange("{$this->prefix}cost:{$queue}", 0, -1);
        $costValues = [];
        foreach ($costs as $cost) {
            $costValues[] = json_decode($cost, true)['cost'];
        }
        sort($costValues);
        $metrics['avg_cost'] = !empty($costValues) ? 
            array_sum($costValues) / count($costValues) : 0;
        $metrics['p95_cost'] = !empty($costValues) ? 
            $costValues[(int)(count($costValues) * 0.95)] : 0;
        // 失败率
        $failed = $this->redis->hGet(
            "{$this->prefix}status:{$queue}:" . floor($now / 60) * 60,
            'failed'
        ) ?: 0;
        $total = $this->redis->hGet(
            "{$this->prefix}status:{$queue}:" . floor($now / 60) * 60,
            'success'
        ) ?: 0;
        $metrics['failure_rate'] = ($total + $failed) > 0 ? 
            $failed / ($total + $failed) * 100 : 0;
        return $metrics;
    }
}
// 使用示例
class JobProcessor
{
    private QueueMonitor $monitor;
    public function handle(Job $job): void
    {
        $startTime = microtime(true);
        try {
            // 处理业务逻辑
            $job->process();
            $this->monitor->recordConsumption('default', [
                'pushed_at' => $job->pushedAt,
                'success' => true
            ]);
        } catch (\Exception $e) {
            $this->monitor->recordConsumption('default', [
                'pushed_at' => $job->pushedAt,
                'success' => false
            ]);
            throw $e;
        }
    }
}

使用 Prometheus + Grafana（推荐生产环境）

// 使用 promphp/prometheus_client_php
use Prometheus\CollectorRegistry;
use Prometheus\Storage\Redis;
class PrometheusMetrics
{
    private CollectorRegistry $registry;
    public function __construct()
    {
        $adapter = new Redis(['host' => '127.0.0.1']);
        $this->registry = new CollectorRegistry($adapter);
    }
    public function initMetrics(): void
    {
        // 队列深度 Gauge
        $gauge = $this->registry->registerGauge(
            'queue',
            'queue_depth',
            'Current queue depth',
            ['queue', 'host']
        );
        // 生产速率 Counter
        $counter = $this->registry->registerCounter(
            'queue',
            'messages_produced_total',
            'Total number of produced messages',
            ['queue']
        );
        // 处理耗时 Histogram
        $histogram = $this->registry->registerHistogram(
            'queue',
            'job_processing_seconds',
            'Job processing duration in seconds',
            ['queue', 'job_type'],
            [0.01, 0.05, 0.1, 0.5, 1, 2, 5]
        );
        // 失败率 Counter
        $failedCounter = $this->registry->registerCounter(
            'queue',
            'failed_jobs_total',
            'Total number of failed jobs',
            ['queue', 'error_type']
        );
    }
    public function recordMetrics(string $queue, array $stats): void
    {
        $gauge = $this->registry->getGauge('queue', 'queue_depth');
        $gauge->set($stats['depth'], [$queue, gethostname()]);
        $counter = $this->registry->getCounter('queue', 'messages_produced_total');
        $counter->incBy($stats['produced_count'], [$queue]);
        if (isset($stats['processing_time'])) {
            $histogram = $this->registry->getHistogram('queue', 'job_processing_seconds');
            $histogram->observe($stats['processing_time'], [$queue, $stats['job_type']]);
        }
        if ($stats['failed']) {
            $failedCounter = $this->registry->getCounter('queue', 'failed_jobs_total');
            $failedCounter->inc([$queue, $stats['error_type'] ?? 'unknown']);
        }
    }
}

实时监控实现

使用 WebSocket 推送实时数据

// 基于 Swoole + WebSocket 的实时监控
class QueueRealTimeMonitor
{
    public function __construct(
        private Redis $redis,
        private \Swoole\WebSocket\Server $ws
    ) {}
    public function broadcastMetrics(): void
    {
        // 每5秒广播一次
        swoole_timer_tick(5000, function () {
            $metrics = $this->collectAllMetrics();
            foreach ($this->ws->connections as $fd) {
                if ($this->ws->isEstablished($fd)) {
                    $this->ws->push($fd, json_encode([
                        'type' => 'metrics',
                        'data' => $metrics
                    ]));
                }
            }
        });
    }
    private function collectAllMetrics(): array
    {
        return [
            'timestamp' => microtime(true),
            'queues' => [
                'default' => $this->getQueueMetrics('default'),
                'high' => $this->getQueueMetrics('high'),
                'email' => $this->getQueueMetrics('email'),
            ],
            'system' => [
                'workers' => $this->getActiveWorkers(),
                'memory_usage' => memory_get_usage(true),
                'cpu_load' => sys_getloadavg()[0],
            ]
        ];
    }
}

日志采集方案

使用结构化日志 + ELK 栈：

class QueueLogger
{
    private Logger $logger;
    public function logMetrics(string $queue, array $context): void
    {
        $this->logger->info('queue.metrics', [
            'queue' => $queue,
            'timestamp' => time(),
            'depth' => $context['depth'],
            'throughput' => $context['throughput'],
            'latency_ms' => $context['latency'] ?? null,
            'worker_id' => getmypid(),
            'host' => gethostname(),
        ]);
    }
}

可视化展示

Grafana Dashboard 配置示例

{: "PHP Queue Monitor",
  "panels": [
    {
      "title": "Queue Depth",
      "type": "graph",
      "targets": [
        {
          "expr": "queue_queue_depth{queue=~\"$queue\"}",
          "legendFormat": "{{queue}} - {{host}}"
        }
      ]
    },
    {
      "title": "Processing Rate",
      "type": "graph",
      "targets": [
        {
          "expr": "rate(queue_messages_produced_total[1m])",
          "legendFormat": "{{queue}}"
        }
      ]
    },
    {
      "title": "P95 Processing Time",
      "type": "heatmap",
      "targets": [
        {
          "expr": "histogram_quantile(0.95, sum(rate(queue_job_processing_seconds_bucket[5m])) by (le, queue))",
          "legendFormat": "{{queue}}"
        }
      ]
    }
  ]
}

最佳实践建议

选择合适存储：
- 实时指标：Redis（Pipeline 批量写入）
- 历史趋势：InfluxDB/TimescaleDB
- 聚合分析：ClickHouse
控制采样频率：
- 高吞吐场景：每5-10秒采样一次
- 低吞吐场景：每60秒采样一次

避免副作用：

// 异步指标上报，不阻塞主流程
$promise = \Amp\call(function () use ($metrics) {
    yield $this->metricsClient->pushAsync($metrics);
});

告警规则：

# prometheus-alert.yml
groups:
- name: queue_alerts
  rules:
  - alert: QueueTooDeep
    expr: queue_queue_depth > 10000
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Queue {{ $labels.queue }} is too deep"
  - alert: HighFailureRate
    expr: rate(queue_failed_jobs_total[5m]) / rate(queue_messages_produced_total[5m]) > 0.05
    for: 3m
    labels:
      severity: critical

实际部署架构

[PHP Workers] → [Redis/Beanstalkd] → [Metrics Collector]
                                            ↓
[Prometheus Pushgateway] ← [Prometheus Client]
                                            ↓
                                      [Prometheus Server]
                                            ↓
                                      [Alertmanager] → [钉钉/邮件/企微]
                                            ↓
                                      [Grafana Dashboard]

这种架构可以同时保证：

低延迟的指标采集（微秒级）
高可用性（无单点故障）
可扩展性（水平扩展 Worker 自动增加指标采集能力）
丰富的可视化（Grafana 社区有成熟的 Queue Dashboard 模板）