Symfony 6.x Messenger同步/异步模式共存配置求助

第一步：更新 messenger.yaml 配置 #

在配置文件中同时定义同步、异步两种传输通道，并设置路由规则区分消息流向：

# config/packages/messenger.yaml
framework:
    messenger:
        default_bus: command.bus
        transports:
            # 同步传输：用于立即发送的事务邮件
            sync: 'sync://'
            # 异步传输：基于Doctrine队列，用于批量邮件
            async: 'doctrine://default'
        routing:
            # 方案1：用不同消息类区分传输（推荐，逻辑清晰）
            App\Message\ImmediateEmailMessage: sync
            App\Message\BulkEmailMessage: async
            
            # 方案2：同一消息类用标记区分（适合不想新增类的场景）
            # App\Message\EmailMessage:
            #     - { transport: sync, when: "message.isImmediate()" }
            #     - { transport: async }

第二步：定义邮件消息类 #

根据选择的方案创建对应的消息载体：

方案1：分拆为两个消息类

// src/Message/ImmediateEmailMessage.php
namespace App\Message;

class ImmediateEmailMessage
{
    public function __construct(
        private string $to,
        private string $subject,
        private string $content
    ) {}

    // 自动生成getter方法（Symfony 6+可使用属性提升省略）
    public function getTo(): string { return $this->to; }
    public function getSubject(): string { return $this->subject; }
    public function getContent(): string { return $this->content; }
}

// src/Message/BulkEmailMessage.php
namespace App\Message;

class BulkEmailMessage
{
    public function __construct(
        private array $recipients,
        private string $subject,
        private string $content
    ) {}

    public function getRecipients(): array { return $this->recipients; }
    public function getSubject(): string { return $this->subject; }
    public function getContent(): string { return $this->content; }
}

方案2：单消息类加同步标记

// src/Message/EmailMessage.php
namespace App\Message;

class EmailMessage
{
    public function __construct(
        private string|array $to,
        private string $subject,
        private string $content,
        private bool $isImmediate = false
    ) {}

    // getter方法
    public function getTo(): string|array { return $this->to; }
    public function getSubject(): string { return $this->subject; }
    public function getContent(): string { return $this->content; }
    public function isImmediate(): bool { return $this->isImmediate; }
}

第三步：编写消息处理器 #

创建对应处理器，调用SendGrid发送邮件：

// src/MessageHandler/ImmediateEmailHandler.php
namespace App\MessageHandler;

use App\Message\ImmediateEmailMessage;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use SendGrid;
use SendGrid\Mail\Mail;

#[AsMessageHandler]
class ImmediateEmailHandler
{
    public function __construct(private SendGrid $sendGrid) {}

    public function __invoke(ImmediateEmailMessage $message): void
    {
        $email = new Mail();
        $email->setFrom('your@domain.com', 'Your App');
        $email->setSubject($message->getSubject());
        $email->addTo($message->getTo());
        $email->addContent('text/html', $message->getContent());

        $this->sendGrid->send($email);
    }
}

// src/MessageHandler/BulkEmailHandler.php
namespace App\MessageHandler;

use App\Message\BulkEmailMessage;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use SendGrid;
use SendGrid\Mail\Mail;

#[AsMessageHandler]
class BulkEmailHandler
{
    public function __construct(private SendGrid $sendGrid) {}

    public function __invoke(BulkEmailMessage $message): void
    {
        foreach ($message->getRecipients() as $recipient) {
            $email = new Mail();
            $email->setFrom('your@domain.com', 'Your App');
            $email->setSubject($message->getSubject());
            $email->addTo($recipient);
            $email->addContent('text/html', $message->getContent());

            $this->sendGrid->send($email);
        }
    }
}

如果用方案2，只需要一个处理器，根据isImmediate标记处理即可（逻辑和上面类似，略作调整）。

第四步：业务代码中发送邮件 #

在控制器/服务中根据需求选择对应消息类发送：

// 发送立即事务邮件（方案1）
use App\Message\ImmediateEmailMessage;
use Symfony\Component\Messenger\MessageBusInterface;

public function confirmOrder(MessageBusInterface $bus)
{
    $bus->dispatch(new ImmediateEmailMessage(
        'user@example.com',
        '订单确认通知',
        '您的订单已成功创建，订单号：XXX'
    ));
}

// 发送批量邮件（方案1）
use App\Message\BulkEmailMessage;
use Symfony\Component\Messenger\MessageBusInterface;

public function sendMonthlyNewsletter(MessageBusInterface $bus)
{
    $recipients = ['user1@example.com', 'user2@example.com', ...];
    $bus->dispatch(new BulkEmailMessage(
        $recipients,
        '月度活动通讯',
        '本月最新活动内容如下：...'
    ));
}

第五步：调整Cron Worker命令 #

确保你的cron.php中执行的Worker仅监听异步传输通道：

php bin/console messenger:consume async --time-limit=300

（结合你已实现的CronRunningEvent空闲自动停止逻辑，无需额外调整）

测试验证 #

1. 发送事务邮件：检查邮箱是否立即收到邮件
2. 发送批量邮件：查看Doctrine的messenger_messages表是否新增记录，执行Worker命令后确认邮件批量发出

内容的提问来源于stack exchange，提问作者user24640806