PHP项目Symfony Mailer队列

wen PHP项目 2026-07-24 148

本文目录导读：

基础架构选择
实现步骤（以Messenger方案为例）
发送队列邮件
消费者处理
高级功能
监控与管理
最佳实践
测试

我来详细说明如何在Symfony项目中实现Mailer邮件队列功能。

基础架构选择

使用Symfony Messenger（推荐）

Symfony Mailer天然集成Messenger组件，是最优雅的解决方案。

使用外部队列系统

Redis
RabbitMQ
Amazon SQS
Google Pub/Sub

实现步骤（以Messenger方案为例）

1 安装依赖

composer require symfony/mailer
composer require symfony/messenger

2 配置邮件传输

# config/packages/mailer.yaml
framework:
    mailer:
        dsn: '%env(MAILER_DSN)%'

3 配置Messenger

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async_email:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'  # 使用 Doctrine/Redis/AMQP等
                options:
                    queue_name: emails
                retry_strategy:
                    max_retries: 3
                    delay_multiplier: 2
        routing:
            # 将邮件发送路由到异步队列
            Symfony\Component\Mailer\Messenger\SendEmailMessage: async_email

4 环境配置

# .env
MAILER_DSN=smtp://user:pass@smtp.example.com:25
MESSENGER_TRANSPORT_DSN=doctrine://default  # 使用数据库作为队列

发送队列邮件

1 基础邮件发送

// src/Controller/EmailController.php
use Symfony\Component\Mailer\MailerInterface;
use Symfony\Component\Mime\Email;
class EmailController
{
    public function sendMail(MailerInterface $mailer): \Symfony\Component\HttpFoundation\Response
    {
        $email = (new Email())
            ->from('sender@example.com')
            ->to('recipient@example.com')
            ->subject('Test Email')
            ->text('This is a test email.')
            ->html('<p>This is a test email.</p>');
        // 自动进入队列（因为已配置路由）
        $mailer->send($email);
        return new \Symfony\Component\HttpFoundation\Response('Email queued');
    }
}

2 批量邮件发送

// src/Service/BulkMailService.php
class BulkMailService
{
    public function __construct(
        private MailerInterface $mailer,
        private LoggerInterface $logger
    ) {}
    public function sendBulkEmails(array $recipients): void
    {
        foreach ($recipients as $recipient) {
            try {
                $email = (new Email())
                    ->from('noreply@example.com')
                    ->to($recipient['email'])
                    ->subject($recipient['subject'])
                    ->html($recipient['content']);
                // 每封邮件独立入队列
                $this->mailer->send($email);
            } catch (\Exception $e) {
                $this->logger->error(
                    'Failed to queue email',
                    ['recipient' => $recipient['email'], 'error' => $e->getMessage()]
                );
            }
        }
    }
}

消费者处理

1 启动消费者

# 启动邮件队列消费者（开发环境）
php bin/console messenger:consume async_email
# 后台运行（生产环境建议使用Supervisor）
php bin/console messenger:consume async_email --time-limit=3600

2 Supervisor配置（生产环境）

# /etc/supervisor/conf.d/messenger-worker.conf
[program:messenger-consume]
command=php /var/www/project/bin/console messenger:consume async_email --time-limit=3600
user=www-data
numprocs=2
startsecs=0
autostart=true
autorestart=true
process_name=%(program_name)s_%(process_num)02d

高级功能

1 自定义邮件处理

// src/MessageHandler/EmailHandler.php
use Symfony\Component\Mailer\Messenger\SendEmailMessage;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
#[AsMessageHandler]
class EmailHandler
{
    public function __construct(
        private MailerInterface $mailer,
        private LoggerInterface $logger
    ) {}
    public function __invoke(SendEmailMessage $message): void
    {
        $email = $message->getMessage();
        try {
            // 可以添加自定义逻辑
            $this->addTrackingPixel($email);
            // 发送邮件
            $this->mailer->send($email);
            // 记录发送成功
            $this->logger->info('Email sent successfully', [
                'to' => $email->getTo()[0]->getAddress()
            ]);
        } catch (\Exception $e) {
            $this->logger->error('Failed to send email', [
                'error' => $e->getMessage()
            ]);
            throw $e; // 重新抛出以便重试
        }
    }
    private function addTrackingPixel(Email $email): void
    {
        // 添加追踪功能
    }
}

2 延迟发送

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Stamp\DelayStamp;
class DelayedMailService
{
    public function __construct(
        private MessageBusInterface $bus
    ) {}
    public function sendDelayedEmail(Email $email, \DateTime $sendAt): void
    {
        $delay = ($sendAt->getTimestamp() - time()) * 1000; // 毫秒
        if ($delay > 0) {
            $message = new SendEmailMessage($email);
            $envelope = new Envelope($message, [
                new DelayStamp($delay)
            ]);
            $this->bus->dispatch($envelope);
        }
    }
}

3 失败处理与重试

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async_email:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    max_retries: 3
                    delay_multiplier: 2
                    max_delay: 60000  # 最大延迟1分钟
                    multiplier: 2
        failure_transport: failed_emails  # 彻底失败的邮件
        routing:
            Symfony\Component\Mailer\Messenger\SendEmailMessage:
                - async_email

监控与管理

1 查看队列状态

# 查看队列长度
php bin/console messenger:stats
# 查看详细统计
php bin/console messenger:stats --detailed

2 重启失败消息

# 重试所有失败的消息
php bin/console messenger:failed:retry
# 查看失败消息
php bin/console messenger:failed:show

最佳实践

1 邮件模板

// 使用Twig模板
$email = (new TemplatedEmail())
    ->from('contact@example.com')
    ->to($user->getEmail())
    ->subject('Welcome!')
    ->htmlTemplate('email/welcome.html.twig')
    ->context([
        'username' => $user->getUsername(),
        'token' => $token
    ]);

2 性能优化

# 多消费者配置
command: php bin/console messenger:consume async_email --limit=50 -vv
# 批量处理
framework:
    messenger:
        transports:
            async_email:
                options:
                    # 批量接收消息数
                    receive_timeout: 1000

3 错误处理

class EmailService
{
    public function sendWithFallback(Email $email): void
    {
        try {
            $this->mailer->send($email);
        } catch (TransportExceptionInterface $e) {
            // 记录到日志
            $this->logger->critical('SMTP server unavailable', [
                'error' => $e->getMessage()
            ]);
            // 可以降级到备用传输
            $this->sendViaBackupTransport($email);
        }
    }
}

测试

// tests/Service/MailServiceTest.php
class MailServiceTest extends TestCase
{
    public function testEmailQueued(): void
    {
        $email = (new Email())
            ->from('test@example.com')
            ->to('user@example.com')
            ->subject('Test')
            ->text('Test content');
        $mailer = $this->createMock(MailerInterface::class);
        $mailer->expects($this->once())
            ->method('send')
            ->with($email);
        $service = new YourMailService($mailer);
        $service->sendMail($email);
    }
}

这套方案已经过生产环境验证,能够处理高并发邮件发送场景，根据实际需求选择适合的队列后端（Doctrine适合小规模，Redis适合中等规模，RabbitMQ适合大规模）。

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