PHP项目死信消息如何监控清理

wen PHP项目 2026-07-01 165

本文目录导读：

PHP项目死信消息如何监控清理

常见死信场景
监控方案
清理策略
健康检查API
最佳实践
日志和审计

在PHP项目中监控和清理死信消息（Dead Letter Messages），通常涉及消息队列系统（如RabbitMQ、Kafka、Redis等），以下是系统的监控和清理方案：

常见死信场景

// 死信常见原因
- 消息被消费者拒绝(Reject)且 requeue=false
- 消息TTL过期
- 队列达到最大长度
- 消费者处理失败次数超限

监控方案

1 使用消息队列管理工具

RabbitMQ 管理界面

# 查看死信队列
rabbitmqctl list_queues name messages messages_unacknowledged messages_ready
# 查看死信交换器
rabbitmqctl list_exchanges name type

2 自定义监控脚本

<?php
class DeadLetterMonitor
{
    private $connection;
    public function __construct()
    {
        $this->connection = new AMQPConnection([
            'host' => 'localhost',
            'port' => 5672,
            'vhost' => '/',
            'login' => 'guest',
            'password' => 'guest'
        ]);
    }
    // 监控死信队列大小
    public function monitorDeadLetterQueue($deadQueueName)
    {
        $channel = new AMQPChannel($this->connection);
        $queue = new AMQPQueue($channel);
        $queue->setName($deadQueueName);
        return [
            'name' => $deadQueueName,
            'message_count' => $queue->declareQueue(),
            'consumer_count' => $queue->getConsumerCount()
        ];
    }
    // 发送告警
    public function checkAndAlert($threshold = 100)
    {
        $metrics = $this->monitorDeadLetterQueue('dead_letter_queue');
        if ($metrics['message_count'] > $threshold) {
            // 发送告警通知
            $this->sendAlert("死信队列消息数: {$metrics['message_count']}");
            // 记录日志
            $this->logDeadLetter($metrics);
        }
    }
    private function sendAlert($message)
    {
        // 实现邮件、短信、Slack等通知
        // mail(), Http client etc.
    }
    private function logDeadLetter($metrics)
    {
        $log = date('Y-m-d H:i:s') . " | " . 
               json_encode($metrics) . PHP_EOL;
        file_put_contents('/var/log/dead_letter_monitor.log', $log, FILE_APPEND);
    }
}

3 使用Prometheus + Grafana

// metrics端点
class DeadLetterMetrics
{
    public function getMetrics()
    {
        $metrics = [];
        // 死信队列大小
        $metrics['dead_letter_queue_size'] = $this->getQueueSize('dead_letter');
        // 死信处理失败次数
        $metrics['dead_letter_process_failures'] = $this->getFailureCount();
        // 死信重试次数分布
        $metrics['dead_letter_retry_distribution'] = $this->getRetryDistribution();
        return $metrics;
    }
}

清理策略

1 自动清理脚本

<?php
class DeadLetterCleaner
{
    private $maxRetries = 3;
    private $retryDelay = 60; // 秒
    // 清理死信消息
    public function cleanDeadLetters()
    {
        try {
            $deadMessages = $this->fetchDeadLetters();
            foreach ($deadMessages as $message) {
                $this->processDeadLetter($message);
            }
        } catch (Exception $e) {
            $this->logError("清理死信失败: " . $e->getMessage());
        }
    }
    // 处理单个死信
    private function processDeadLetter($message)
    {
        $messageId = $message['message_id'];
        $retryCount = $message['retry_count'] ?? 0;
        // 1. 记录死信详情
        $this->logDeadLetterDetail($message);
        // 2. 判断重试次数
        if ($retryCount < $this->maxRetries) {
            // 重新投递到原始队列
            $this->reQueueMessage($message);
        } else {
            // 3. 超过重试次数，归档或丢弃
            $this->archiveDeadLetter($message);
            // 或直接删除
            $this->deleteDeadLetter($message);
        }
        // 4. 正式确认消费（ACK）
        $this->acknowledgeMessage($messageId);
    }
    // 重新入队
    private function reQueueMessage($message)
    {
        $channel = new AMQPChannel($this->connection);
        $exchange = new AMQPExchange($channel);
        $exchange->setName('original_exchange');
        // 修改消息属性，增加重试计数
        $properties = $message['properties'];
        $properties['retry_count'] = ($message['retry_count'] ?? 0) + 1;
        $properties['last_retry_time'] = time();
        // 重新投递
        $exchange->publish(
            $message['body'],
            $message['routing_key'],
            AMQP_NOPARAM,
            $properties
        );
    }
    // 归档死信
    private function archiveDeadLetter($message)
    {
        // 存储到数据库或文件系统
        $archive = [
            'message_id' => $message['message_id'],
            'body' => $message['body'],
            'headers' => $message['headers'],
            'failed_at' => date('Y-m-d H:i:s'),
            'retry_count' => $message['retry_count']
        ];
        // 保存到MySQL
        DB::table('dead_letter_archive')->insert($archive);
        // 或保存到文件
        file_put_contents(
            "/var/dead_letters/{$message['message_id']}.json",
            json_encode($archive)
        );
    }
    // 批量清理过期死信
    public function cleanExpiredDeadLetters($expireHours = 24)
    {
        $cutoffTime = time() - ($expireHours * 3600);
        // 删除过期死信
        DB::table('dead_letter_archive')
            ->where('created_at', '<', date('Y-m-d H:i:s', $cutoffTime))
            ->delete();
        // 清理日志文件
        $this->cleanExpiredLogs($expireHours);
    }
}

2 定时任务清理

# crontab 配置
# 每分钟检查死信队列
* * * * * php /path/to/check_dead_letters.php
# 每5分钟清理过期死信
*/5 * * * * php /path/to/clean_dead_letters.php
# 每小时汇报死信统计
0 * * * * php /path/to/report_dead_letters.php

3 基于Redis的清理

<?php
class RedisDeadLetterCleaner
{
    private $redis;
    public function __construct()
    {
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
    }
    // 清理Redis死信
    public function cleanRedisDeadLetters()
    {
        // 1. 获取所有死信key
        $deadKeys = $this->redis->keys('dead_letter:*');
        foreach ($deadKeys as $key) {
            $deadData = $this->redis->hGetAll($key);
            // 检查TTL
            $ttl = $this->redis->ttl($key);
            if ($ttl < 0) {
                // 设置过期时间
                $this->redis->expire($key, 86400); // 24小时
            }
            // 检查重试次数
            $retryCount = $deadData['retry_count'] ?? 0;
            if ($retryCount >= 3) {
                // 移动到归档列表
                $this->redis->rPush('dead_letter_archive', json_encode($deadData));
                // 删除原始key
                $this->redis->del($key);
            }
        }
    }
    // 获取死信统计
    public function getDeadLetterStats()
    {
        return [
            'total' => $this->redis->lLen('dead_letter_archive'),
            'recent' => $this->redis->lRange('dead_letter_archive', 0, 10)
        ];
    }
}

健康检查API

<?php
// health_check.php
class DeadLetterHealthCheck
{
    public function check()
    {
        $status = [
            'healthy' => true,
            'dead_letter_queue_size' => 0,
            'message' => 'OK'
        ];
        try {
            // 检查死信队列
            $queueSize = $this->getDeadLetterQueueSize();
            $status['dead_letter_queue_size'] = $queueSize;
            if ($queueSize > 1000) {
                $status['healthy'] = false;
                $status['message'] = "死信队列积压: {$queueSize}条";
            }
            // 检查最近处理成功率
            $successRate = $this->getRecentSuccessRate();
            if ($successRate < 0.95) {
                $status['healthy'] = false;
                $status['message'] .= ", 处理成功率: {$successRate}%";
            }
        } catch (Exception $e) {
            $status['healthy'] = false;
            $status['message'] = "检查失败: " . $e->getMessage();
        }
        return $status;
    }
    private function getDeadLetterQueueSize()
    {
        // 实现具体队列大小获取逻辑
        return 50;
    }
    private function getRecentSuccessRate()
    {
        // 计算最近5分钟的处理成功率
        return 0.98;
    }
}
// 返回健康检查结果
header('Content-Type: application/json');
$checker = new DeadLetterHealthCheck();
echo json_encode($checker->check());

最佳实践

1 死信处理流程

// 1. 定义重试策略
$retryPolicy = [
    'max_retries' => 3,
    'backoff_strategy' => 'exponential', // 指数退避
    'initial_delay' => 5, // 秒
    'multiplier' => 2
];
// 2. 死信分类处理
class DeadLetterClassifier
{
    public function classify($message)
    {
        $errorCode = $message['headers']['error_code'] ?? 0;
        return match(true) {
            $errorCode >= 500 => 'system_error', // 系统错误，可重试
            $errorCode >= 400 => 'client_error', // 客户端错误，不可重试
            $message['headers']['timeout'] ?? false => 'timeout_error', // 超时
            default => 'unknown'
        };
    }
}
// 3. 自动修复机制
class AutoRepairService
{
    public function attemptAutoRepair($deadLetter)
    {
        $classifier = new DeadLetterClassifier();
        $type = $classifier->classify($deadLetter);
        return match($type) {
            'system_error' => $this->retryWithBackoff($deadLetter),
            'client_error' => $this->notifyAdmin($deadLetter),
            'timeout_error' => $this->increaseTimeout($deadLetter),
            default => $this->logAndArchive($deadLetter)
        };
    }
}

2 预警配置

// 预警阈值配置
$alertConfig = [
    'critical_threshold' => 1000,  // 严重
    'warning_threshold' => 500,    // 警告
    'info_threshold' => 100,       // 信息
    // 监控周期
    'check_interval' => 60, // 秒
    // 通知渠道
    'channels' => [
        'email' => ['admin@example.com'],
        'slack' => '#alerts',
        'webhook' => 'https://hooks.example.com/dead_letter'
    ]
];

日志和审计

// 死信审计日志
class DeadLetterAudit
{
    public function logDeadLetterEvent($event, $details)
    {
        $logEntry = [
            'timestamp' => date('Y-m-d H:i:s'),
            'event' => $event, // created, processed, archived, deleted
            'details' => $details,
            'source_ip' => $_SERVER['REMOTE_ADDR'] ?? 'unknown',
            'user_agent' => $_SERVER['HTTP_USER_AGENT'] ?? 'unknown'
        ];
        // 写入日志
        $this->writeAuditLog($logEntry);
        // 同步到监控系统
        $this->syncToMonitor($logEntry);
    }
}

一个完善的PHP死信监控清理系统应该：