本文目录导读：
我来详细说明如何在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适合大规模）。
