PHP项目邮件队列与发送限额

wen PHP项目 2026-07-21 156

本文目录导读：

PHP项目邮件队列与发送限额

整体架构设计
核心代码实现
进阶优化
性能监控与告警
最佳实践建议
完整项目结构示例

针对PHP项目的邮件队列与发送限额问题,通常是为了解决以下核心痛点：

高并发下API限流：SMTP服务商（如SendGrid、阿里云邮件）通常有每秒/每小时的发送上限。
慢速IO阻塞：发送邮件是网络I/O操作,同步发送会拖慢用户请求响应时间。
失败重试：临时性发送失败（如服务商返回429 Too Many Requests）需要自动重试。

以下是基于 PHP + Redis + Supervisor 的成熟解决方案架构与代码示例。

整体架构设计

[Web Request] -> [Queue Push] -> [Redis List] -> [Worker进程] -> [限流器] -> [SMTP Service]
                                                                          -> [失败重入队列]

队列存储：Redis（PHP常用），也可用 RabbitMQ / Beanstalkd。
Worker：PHP CLI 脚本，常驻运行（由 Supervisor 管理）。
限流机制：基于 Redis + Lua 实现令牌桶或滑动窗口。

核心代码实现

封装邮件投递类（DSendMail）

<?php
// MailSender.php
use Predis\Client as RedisClient;
class MailSender {
    private RedisClient $redis;
    private string $queueKey = 'email:queue';
    // 限流参数
    private int $maxPerSecond = 5;   // 每秒最多5封
    private int $maxPerHour  = 1000; // 每小时最多1000封
    public function __construct() {
        $this->redis = new RedisClient([
            'scheme' => 'tcp',
            'host'   => '127.0.0.1',
            'port'   => 6379,
        ]);
    }
    /**
     * 将邮件任务推入队列
     */
    public function dispatch(string $to, string $subject, string $body, array $options = []): bool {
        $task = [
            'to'      => $to,
            'subject' => $subject,
            'body'    => $body,
            'options' => $options,
            'created_at' => time(),
            'retry_count' => 0,
        ];
        // 左推入队列（LPUSH）
        $this->redis->lpush($this->queueKey, json_encode($task));
        return true;
    }
    /**
     * Worker 核心：消费队列，带限流
     */
    public function consume(): void {
        while (true) {
            // 阻塞获取任务（BRPOP 无任务时阻塞）
            $taskJson = $this->redis->brpop($this->queueKey, 5);
            if (!$taskJson) continue;
            $task = json_decode($taskJson[1], true);
            // 等待限流许可
            $this->waitForRateLimit();
            // 执行发送
            $result = $this->sendMail($task);
            if ($result['success']) {
                echo "Sent to {$task['to']}\n";
            } else {
                // 失败重试逻辑
                $this->handleFailure($task, $result['error']);
            }
        }
    }
    /**
     * 基于 Redis 的令牌桶限流器
     */
    private function waitForRateLimit(): void {
        $luaScript = <<<LUA
            local key = KEYS[1]
            local maxTokens = tonumber(ARGV[1])  -- 每秒令牌数
            local bucketTime = tonumber(ARGV[2]) -- 令牌桶时间窗口（秒）
            local current = redis.call('GET', key) or 0
            if current < maxTokens then
                redis.call('INCR', key)
                redis.call('EXPIRE', key, bucketTime)
                return 1  -- 允许发送
            else
                return 0  -- 限流
            end
LUA;
        // 秒级限流：每秒最多5封
        while (true) {
            $allowed = $this->redis->eval(
                $luaScript,
                1,
                'email:rate:second',
                5,
                1
            );
            if ($allowed) break;
            usleep(200000); // 等待200ms
        }
        // 小时级限流（类似原理）
        // ...
    }
    private function sendMail(array $task): array {
        try {
            // 调用你的邮件发送逻辑（PHPMailer / Symfony Mailer / Guzzle API）
            // 示例使用 Symfony Mailer
            // $email = (new Email())->to($task['to'])->subject($task['subject'])->html($task['body']);
            // $mailer->send($email);
            return ['success' => true];
        } catch (\Exception $e) {
            return ['success' => false, 'error' => $e->getMessage()];
        }
    }
    private function handleFailure(array $task, string $error): void {
        $maxRetry = 3;
        if ($task['retry_count'] < $maxRetry) {
            $task['retry_count']++;
            // 指数退避：延迟重试时间递增（秒）
            $delay = pow(2, $task['retry_count']); 
            // 推入延迟队列（可利用 Redis 的有序集合实现延迟队列）
            $this->redis->zadd('email:delay', time() + $delay, json_encode($task));
        } else {
            // 记录失败日志到数据库或日志文件
            file_put_contents('/var/log/email_failure.log', json_encode($task) . PHP_EOL, FILE_APPEND);
        }
    }
}

Worker 启动脚本（CLI）

<?php
// worker.php
require 'vendor/autoload.php';
$sender = new MailSender();
echo "Email Worker started...\n";
$sender->consume();

Supervisor 配置

确保 Worker 进程崩溃后自动重启。

[program:email_worker]
command=php /path/to/worker.php
directory=/path/to/project
autostart=true
autorestart=true
user=www-data
numprocs=3           ; 启动3个进程同时消费
process_name=%(program_name)s_%(process_num)02d
stdout_logfile=/var/log/email_worker.log
stderr_logfile=/var/log/email_worker_error.log

进阶优化

使用更专业的限流库：`nikic/php-rate-limiter`

安装：composer require nikic/php-rate-limiter

use RateLimiter\Limiter;
use RateLimiter\Storage\RedisStorage;
$limiter = new Limiter(new RedisStorage($redisClient));
// 每秒5封，突发可10封
$limiter->allow('email:rate', 5, 10, 1);

基于 Laravel 的队列方案（开箱即用）

如果你的项目是 Laravel,可以直接用内置队列系统：

// 1. 配置 .env 使用 Redis 驱动
// QUEUE_CONNECTION=redis
// 2. 创建任务类
php artisan make:job SendEmailJob
// 3. Job 代码
public function handle()
{
    // 限流中间件
    RateLimiter::for('email', function () {
        return Limit::perSecond(5)->perHour(1000);
    });
    Mail::to($this->user)->send($this->email);
}
// 4. 调度任务
SendEmailJob::dispatch($user, $email)->onQueue('emails');

异常重试策略

// 使用 Laravel 的重试机制
public $tries = 3;  // 最多重试3次
public $backoff = [2, 10, 30]; // 重试间隔（秒）
// 或自定义重试逻辑
public function retryUntil(): DateTime
{
    return now()->addMinutes(5); // 5分钟内重试
}

性能监控与告警

队列深度监控：redis-cli LLEN email:queue
告警阈值：队列超过 10,000 封。
发送成功率：记录到 Prometheus / Grafana，email_sent_total{status="success/fail"}。
限流触发次数：email_rate_limit_hit_total。
Worker 状态：supervisorctl status email_worker:*。

最佳实践建议

不要阻塞主请求：用户点击“发送验证码”后立即返回“发送中”，Worker 异步处理。
区分优先级：验证码类邮件高优先级（独立队列）,营销邮件低优先级。
冷却时间：同一收件人发送间隔不少于 30 秒,避免被封。
日志完整：记录每封邮件的 Message ID、发送时间、耗时、错误原因。
邮箱验证：发送前验证邮箱格式（filter_var($email, FILTER_VALIDATE_EMAIL)）。

完整项目结构示例

/src
  /MailSender.php       # 队列推送 + 消费
  /RateLimiter.php      # 令牌桶实现
  /FailureHandler.php   # 失败重试逻辑
/worker.php             # 入口脚本
/config/supervisor.conf # 守护进程配置
/scripts
  /monitor.php          # 队列深度监控
  /retry_failed.php     # 手动重试失败任务

通过以上方案，你可以轻松实现 高吞吐、低延迟、防封禁 的邮件发送系统，核心在于：异步队列解耦 + 限流保护 + 指数退避重试。

本文地址： https://www.bmjup.com/post/26693.html