Symfony Messenger: คิว, Worker และสถาปัตยกรรม Async สำหรับสัมภาษณ์งาน 2026

Symfony Messenger เป็นคอมโพเนนต์หลักสำหรับจัดการงานแบบอะซิงโครนัสในระบบนิเวศ PHP สมัยใหม่ คอมโพเนนต์นี้มอบ message bus ที่มีโครงสร้างชัดเจน, transport ที่กำหนดค่าได้หลากหลาย และ worker ที่ถูกควบคุมอย่างรัดกุม ด้วย Symfony 7.3 และแผนงานสู่เวอร์ชัน 8.0 ฟีเจอร์ใหม่อย่าง middleware ป้องกันข้อความซ้ำ, streaming AMQP และ Doctrine keepalive ทำให้ Messenger มีความสามารถเทียบเท่ากับระบบคิวเฉพาะทางอย่าง Laravel Horizon หรือ Sidekiq

สถาปัตยกรรม Symfony Messenger: Bus, Transport และ Worker

คอมโพเนนต์ Messenger แยกความรับผิดชอบออกเป็นสามส่วน: การส่ง (bus), การขนส่ง (transport) และการประมวลผล (worker) ข้อความ (message) คือออบเจกต์ PHP ธรรมดา Handler คือคลาสที่สามารถเรียกใช้งานได้ (invokable) Bus เชื่อมต่อทั้งสองส่วน โดยกำหนดเส้นทางผ่าน middleware และเลือกที่จะ serialize ข้อความไปยัง transport เพื่อประมวลผลแบบอะซิงโครนัส

namespace App\Message;

final readonly class InvoiceGenerated
{
    public function __construct(
        public int $orderId,
        public string $customerEmail,
    ) {}
}

ข้อความนี้บรรจุเฉพาะข้อมูลแบบ scalar เท่านั้น ไม่ใช่ entity ของ Doctrine การส่ง ID แทนออบเจกต์เต็มรูปแบบช่วยหลีกเลี่ยงปัญหา serialization และทำให้ข้อความเบา Handler จะดึงข้อมูลล่าสุดจากฐานข้อมูล ณ เวลาที่ประมวลผล

namespace App\MessageHandler;

use App\Message\InvoiceGenerated;
use App\Service\InvoiceService;
use App\Service\MailerService;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final readonly class InvoiceGeneratedHandler
{
    public function __construct(
        private InvoiceService $invoiceService,
        private MailerService $mailerService,
    ) {}

    public function __invoke(InvoiceGenerated $message): void
    {
        $pdf = $this->invoiceService->generatePdf($message->orderId);
        $this->mailerService->sendInvoice(
            $message->customerEmail,
            $pdf,
        );
    }
}

การ dispatch ข้อความจาก controller หรือ service ใช้เพียงบรรทัดเดียว:

$this->bus->dispatch(new InvoiceGenerated(
    orderId: $order->getId(),
    customerEmail: $order->getCustomer()->getEmail(),
));

Bus จะตัดสินใจว่าจะทำงานแบบซิงโครนัสหรืออะซิงโครนัสตามการกำหนดค่า routing ของ transport

การกำหนดค่า Transport และ Backend คิว

Messenger รองรับหลาย backend: Doctrine DBAL, Redis, Amazon SQS, Beanstalkd, AMQP (RabbitMQ) และ streaming AMQP transport ที่เปิดตัวในปี 2025 แต่ละ transport ถูกกำหนดด้วย DSN string

# config/packages/messenger.yaml
framework:
    messenger:
        failure_transport: failed

        transports:
            async_priority_high:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: high
                retry_strategy:
                    max_retries: 3
                    delay: 1000
                    multiplier: 3
                    max_delay: 60000

            async_priority_low:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: low
                retry_strategy:
                    max_retries: 5
                    delay: 5000
                    multiplier: 2

            failed:
                dsn: 'doctrine://default?queue_name=failed'

        routing:
            'App\Message\InvoiceGenerated': async_priority_high
            'App\Message\CleanupTempFiles': async_priority_low

การแบ่ง transport ตามลำดับความสำคัญช่วยให้งานที่ต้องตอบสนองรวดเร็ว (สร้างใบแจ้งหนี้) ถูกประมวลผลก่อนงานบำรุงรักษา (ลบไฟล์ชั่วคราว) แต่ละ transport มีกลยุทธ์ retry ของตัวเอง ซึ่งปรับแต่งตามระดับความสำคัญและคุณสมบัติ idempotent ของข้อความ

การจัดการ Worker ด้วย Supervisor

Worker รับข้อความจาก transport ในสภาพแวดล้อม production คำสั่ง messenger:consume จะทำงานภายใต้ process manager อย่าง Supervisor หรือ systemd

# รับข้อความลำดับความสำคัญสูงก่อน จากนั้นลำดับความสำคัญต่ำ
php bin/console messenger:consume async_priority_high async_priority_low \
    --memory-limit=128M \
    --time-limit=3600 \
    --limit=500

flag จำกัดทั้งสามตัวนี้ป้องกันการรั่วไหลของหน่วยความจำและทำให้ worker รีสตาร์ทเป็นระยะ Supervisor จะเริ่มกระบวนการใหม่โดยอัตโนมัติหลังจากออกแต่ละครั้ง

; /etc/supervisor/conf.d/messenger-worker.conf
[program:messenger-consume]
command=php /var/www/app/bin/console messenger:consume async_priority_high async_priority_low --memory-limit=128M --time-limit=3600
user=www-data
numprocs=2
process_name=%(program_name)s_%(process_num)02d
autostart=true
autorestart=true
startsecs=0
stopwaitsecs=30
stdout_logfile=/var/log/messenger-worker.log
stderr_logfile=/var/log/messenger-worker-error.log

การตั้งค่า numprocs=2 จะสร้าง worker แบบขนานสองตัว ทำให้ throughput เพิ่มเป็นสองเท่า สามารถปรับจำนวนตาม core CPU ของเซิร์ฟเวอร์และเวลาประมวลผลข้อความ

Pipeline Middleware และ CQRS ด้วยหลาย Bus

Middleware ครอบคลุมทุกการ dispatch ข้อความ เพิ่ม cross-cutting concern ต่างๆ stack middleware ในตัวจัดการ validation, Doctrine transaction และ routing

# config/packages/messenger.yaml
framework:
    messenger:
        default_bus: command.bus
        buses:
            command.bus:
                middleware:
                    - validation
                    - doctrine_transaction
            query.bus:
                middleware:
                    - validation
            event.bus:
                default_middleware:
                    allow_no_handlers: true
                middleware:
                    - validation

การกำหนดค่านี้นำรูปแบบ CQRS (Command Query Responsibility Segregation) มาใช้ Command เปลี่ยนแปลง state ภายใน Doctrine transaction Query อ่านข้อมูลอย่างเดียว Event อนุญาตให้ไม่มี handler เลยก็ได้ ทำให้สามารถใช้รูปแบบ pub/sub ที่ listener สามารถเพิ่มเข้ามาได้อย่างอิสระ

Middleware ป้องกันข้อความซ้ำใน Symfony 7.3

ข้อความซ้ำสิ้นเปลืองทรัพยากรและอาจก่อให้เกิดผลข้างเคียง เช่น เรียกเก็บเงินลูกค้าสองครั้ง Symfony 7.3 เปิดตัว DeduplicateMiddleware เพื่อข้ามข้อความที่เหมือนกันซึ่งอยู่ในคิวแล้วโดยอัตโนมัติ

namespace App\Message;

use Symfony\Component\Messenger\Stamp\DeduplicateStamp;

final readonly class SendWelcomeEmail
{
    public function __construct(
        public int $userId,
    ) {}
}

// Dispatch พร้อมการป้องกันข้อความซ้ำ
use Symfony\Component\Messenger\Stamp\DeduplicateStamp;

$this->bus->dispatch(
    new SendWelcomeEmail(userId: 42),
    [new DeduplicateStamp(id: 'welcome-email-42')],
);

DeduplicateStamp รับตัวระบุ lock resource หากข้อความที่มี ID เดียวกันอยู่ในสถานะ pending อยู่แล้ว การ dispatch ใหม่จะถูกข้ามไปโดยไม่แจ้งเตือน ฟีเจอร์นี้ต้องใช้คอมโพเนนต์ Lock ร่วมกับ store ที่ serialize ได้ (Redis, Memcached หรือฐานข้อมูล)

กลยุทธ์ Retry และ Failure Transport

เมื่อ handler โยน exception, Messenger จะลองส่งข้อความซ้ำตามกลยุทธ์ retry ของ transport หลังจากหมดจำนวนครั้งที่กำหนด ข้อความจะถูกย้ายไปยัง failure transport

namespace App\MessageHandler;

use App\Message\ProcessPayment;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Messenger\Exception\RecoverableMessageHandlingException;

#[AsMessageHandler]
final class PaymentHandler
{
    public function __invoke(ProcessPayment $message): void
    {
        try {
            $this->gateway->charge($message->amount, $message->token);
        } catch (GatewayTimeoutException $e) {
            // กู้คืนได้: ลองใหม่ด้วย backoff
            throw new RecoverableMessageHandlingException(
                'Payment gateway timeout, retrying',
                previous: $e,
            );
        } catch (InvalidCardException $e) {
            // กู้คืนไม่ได้: ส่งไป failure transport ทันที
            throw new UnrecoverableMessageHandlingException(
                'Invalid card, no retry',
                previous: $e,
            );
        }
    }
}

RecoverableMessageHandlingException เรียกใช้กลยุทธ์ retry ส่วน UnrecoverableMessageHandlingException ข้ามการ retry ทั้งหมดและส่งข้อความตรงไปยัง failure transport การแยกแยะนี้ช่วยไม่ให้เสียจำนวนครั้ง retry กับข้อความที่ล้มเหลวอย่างถาวร

# ตรวจสอบและจัดการข้อความที่ล้มเหลว
php bin/console messenger:failed:show
php bin/console messenger:failed:show 20 --transport=failed

# ลองส่งข้อความที่ระบุอีกครั้ง
php bin/console messenger:failed:retry 20 30

# กรองและลบตามคลาส (Symfony 7.3+)
php bin/console messenger:failed:remove --class-filter="App\Message\CleanupTempFiles"

Doctrine Keepalive และข้อความที่ใช้เวลานาน

Handler ที่ประมวลผลข้อความเป็นเวลานานเสี่ยงที่จะถูกส่งซ้ำเมื่อ visibility timeout ของ transport หมดอายุ Symfony 7.2 เปิดตัว keepalive สำหรับ Redis, SQS และ Beanstalkd ส่วน Symfony 7.3 ขยายฟีเจอร์นี้ไปยัง transport Doctrine

# เปิด keepalive เพื่อป้องกันการส่งซ้ำระหว่างการประมวลผลที่ใช้เวลานาน
php bin/console messenger:consume async --keepalive

flag --keepalive จะอัปเดต timestamp delivered_at ในตาราง transport ของ Doctrine เป็นระยะ เพื่อแจ้งว่า worker ยังคงประมวลผลข้อความอยู่ หากไม่ใช้ flag นี้ ข้อความที่ประมวลผลนานกว่า timeout ของ transport (ค่าเริ่มต้น 5 นาที) จะถูก worker อื่นนำไปประมวลผล ทำให้เกิดการทำงานซ้ำซ้อน

Attribute #[AsMessage] สำหรับ Routing แบบประกาศ

Symfony 7.2 เปิดตัว attribute #[AsMessage] ซึ่งย้ายการกำหนด transport routing จาก YAML ไปยังคลาส message โดยตรง

namespace App\Message;

use Symfony\Component\Messenger\Attribute\AsMessage;

#[AsMessage(transport: 'async_priority_low')]
final readonly class GenerateReport
{
    public function __construct(
        public int $reportId,
        public string $format = 'pdf',
    ) {}
}

วิธีนี้ไม่ต้องกำหนดส่วน routing ใน messenger.yaml สำหรับข้อความนั้นอีกต่อไป Transport ถูกประกาศที่ต้นทาง ทำให้ codebase อธิบายตัวเองได้ชัดเจน ทั้งสองวิธี (routing YAML และ attribute) สามารถใช้ร่วมกันได้ โดย attribute จะมีลำดับความสำคัญสูงกว่า

Streaming AMQP Transport สำหรับคิว Throughput สูง

Transport AMQP แบบดั้งเดิมใช้ polling (get()) เพื่อดึงข้อความ ซึ่งสร้างภาระที่ไม่จำเป็นบน RabbitMQ ส่วน streaming AMQP transport ที่เปิดตัวในปี 2025 เปลี่ยนเป็นโมเดล push (consume()) ซึ่งลดความหน่วงและการใช้ทรัพยากร

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            streaming:
                dsn: 'amqp-lib://guest:guest@localhost:5672/%2f/messages'
                options:
                    exchange:
                        name: app_events
                        type: topic
                    queues:
                        order_events:
                            binding_keys: ['order.*']

ความแตกต่างหลักจาก transport AMQP เริ่มต้น: ไม่ต้องใช้ C extension (ใช้ php-amqplib), การส่งแบบ streaming ผ่านการเชื่อมต่อ TCP ที่คงอยู่ยาวนาน และรองรับ topic exchange พร้อม routing binding key แบบ native Transport นี้สามารถจัดการข้อความหลายพันรายการต่อวินาทีด้วย CPU overhead ที่น้อยมาก

สรุป

• ส่ง ID แบบ scalar ในข้อความ ไม่ใช่ entity ของ Doctrine; ดึงข้อมูลล่าสุดใน handler เพื่อหลีกเลี่ยงปัญหา serialization และ state ที่ล้าสมัย
• แบ่ง transport ตามลำดับความสำคัญและระดับวิกฤต; แต่ละ transport มีกลยุทธ์ retry และ pool worker ของตัวเอง
• ใช้ RecoverableMessageHandlingException และ UnrecoverableMessageHandlingException เพื่อควบคุมพฤติกรรม retry อย่างชัดเจน
• เปิด --keepalive บน worker transport Doctrine เพื่อป้องกันการส่งซ้ำของข้อความที่ใช้เวลานาน
• ใช้ DeduplicateStamp กับข้อความที่การประมวลผลซ้ำก่อให้เกิดผลข้างเคียง (การชำระเงิน, อีเมล, การแจ้งเตือน)
• นำ CQRS มาใช้ด้วยหลาย bus: command bus สำหรับ mutation พร้อม Doctrine transaction, query bus สำหรับอ่านข้อมูล, event bus สำหรับ pub/sub
• ใช้ Supervisor หรือ systemd ใน production พร้อม flag --memory-limit, --time-limit และ --limit สำหรับจัดการวงจรชีวิต worker
• พิจารณา streaming AMQP transport สำหรับสถานการณ์ throughput สูงที่ต้องการความหน่วงต่ำกว่ามิลลิวินาที