المعالجة غير المتزامنة مع Symfony Messenger

Symfony Messenger: المعالجة غير المتزامنة بكل بساطة

يتيح مكوّن Messenger في Symfony فصل المعالجات الطويلة عن طلبات HTTP، لتحسين تجربة المستخدم.

إرسال بريد إلكتروني، أو توليد ملف PDF، أو استدعاء API خارجية بطيئة، أو تغيير أبعاد صورة: لا تحتاج أيّ من هذه المهام إلى تعطيل الاستجابة المرسَلة إلى المستخدم. مع Messenger، يحفظ المتحكّم (controller) ما هو أساسي، وينشر رسالة في طابور انتظار (queue)، ثم يستجيب فوراً. بعد ذلك تتولّى عملية مستقلة، هي العامل (worker)، معالجة الرسالة في الخلفية.

مكوّنات Messenger

• الرسالة (message): كائن PHP بسيط يحمل البيانات اللازمة للمعالجة، ولا يحتوي على أي منطق.
• المعالِج (handler): الصنف الذي يعرف كيف يعالج نوعاً معيّناً من الرسائل.
• الناقل (bus): نقطة الدخول MessageBusInterface، التي نسلّمها الرسائل عبر dispatch().
• وسيلة النقل (transport): الطابور الذي تنتظر فيه الرسائل (RabbitMQ، Redis، جدول Doctrine، Amazon SQS…).
• العامل (worker): الأمر messenger:consume، الذي يقرأ وسيلة النقل ويستدعي المعالِجات.

من دون إعداد للتوجيه (routing)، تُعالَج الرسالة بشكل متزامن داخل الطلب نفسه. إن توجيهها إلى وسيلة نقل هو ما يجعل المعالجة غير متزامنة، دون المساس بشيفرة الرسالة أو المعالِج.

بنية Message/Handler

// Message
class SendNotificationMessage
{
    public function __construct(
        public readonly int $userId,
        public readonly string $content,
    ) {}
}

// Handler
#[AsMessageHandler]
class SendNotificationHandler
{
    public function __construct(
        private NotificationService $notificationService,
    ) {}

    public function __invoke(SendNotificationMessage $message): void
    {
        $this->notificationService->send(
            $message->userId,
            $message->content,
        );
    }
}

الرسالة غير قابلة للتعديل (readonly) ولا تحتوي إلا على قيم بسيطة (scalars): إذ ستُسلسَل (serialize) لتخزينها في الطابور، ثم يعيد العامل بناءها، أحياناً بعد عدة دقائق. مرّر مُعرّفاً بدلاً من كيان Doctrine: سيعيد المعالِج تحميل نسخة حديثة من قاعدة البيانات. تكفي السمة #[AsMessageHandler] ليربط Symfony المعالِج بالرسالة، بالاعتماد على نوع وسيط الدالة __invoke().

إعداد وسائل النقل

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

        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    max_retries: 3
                    delay: 1000
                    multiplier: 2
            failed:
                dsn: 'doctrine://default?queue_name=failed'

        routing:
            App\Message\SendNotificationMessage: async
            App\Message\ProcessOrderMessage: async

تُعبَّر استراتيجية إعادة المحاولة بالمللي ثانية: عند حدوث استثناء، يُعاد تنفيذ الرسالة بعد ثانية واحدة، ثم ثانيتين، ثم أربع. بعد الإخفاق الثالث، تنتقل إلى وسيلة النقل المحدّدة في failure_transport، وهي هنا جدول Doctrine. من دون هذا المفتاح، تضيع ببساطة كل رسالة استنفدت محاولاتها.

يعتمد DSN وسيلة النقل الرئيسية على بنيتك التحتية:

# .env
# RabbitMQ (requires the amqp PHP extension)
MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages
# Redis
# MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
# Database, no extra infrastructure
# MESSENGER_TRANSPORT_DSN=doctrine://default?auto_setup=0

تُعدّ وسيلة النقل Doctrine نقطة انطلاق جيدة: فهي لا تتطلب شيئاً غير قاعدة البيانات الموجودة. ويصبح RabbitMQ أو Redis خياراً مجدياً حين يزداد حجم الرسائل أو حين تتشارك عدة تطبيقات الطوابير نفسها.

إرسال الرسائل (dispatch)

class OrderController extends AbstractController
{
    public function __construct(
        private OrderService $orderService,
    ) {}

    #[Route('/order', methods: ['POST'])]
    public function create(
        MessageBusInterface $bus,
        Request $request,
    ): JsonResponse {
        // Fast synchronous work
        $order = $this->orderService->create($request);

        // Asynchronous work
        $bus->dispatch(new SendNotificationMessage(
            $order->getUserId(),
            "Order #{$order->getId()} confirmed"
        ));

        return $this->json($order, 201);
    }
}

يبقى إنشاء الطلب متزامناً: يجب أن يعرف المستخدم فوراً هل نجح أم لا. وحده الإشعار يذهب إلى الطابور. أرسل الرسالة بعد الحفظ في قاعدة البيانات: فإذا عالجها العامل قبل انتهاء المعاملة (transaction)، فلن يجد الطلب.

تشغيل العمّال ومراقبتهم

في بيئة التطوير، تكفي نافذة طرفية واحدة:

php bin/console messenger:consume async -vv

# In production: restart the worker regularly
php bin/console messenger:consume async --time-limit=3600 --memory-limit=256M

العامل في PHP عملية طويلة الأمد: لا يعيد قراءة الشيفرة بعد النشر، وقد تتضخّم ذاكرته. يوقفه الخياران --time-limit و--memory-limit بشكل نظيف بعد الرسالة الجارية، فيعيد مدير العمليات تشغيله فوراً. مع systemd، تتيح وحدة قالب (template unit) تشغيل عدة عمّال:

# /etc/systemd/system/messenger-worker@.service
[Unit]
Description=Symfony Messenger worker %i
After=network.target

[Service]
User=app
WorkingDirectory=/var/www/app
ExecStart=/usr/bin/php bin/console messenger:consume async --time-limit=3600 --memory-limit=256M
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable --now messenger-worker@1 messenger-worker@2

عند كل نشر، نفّذ php bin/console messenger:stop-workers: ينهي العمّال رسالتهم الجارية ثم يتوقفون، ويعيد systemd تشغيلهم بالشيفرة الجديدة.

التعامل مع الرسائل الفاشلة

php bin/console messenger:failed:show
php bin/console messenger:failed:retry
php bin/console messenger:failed:remove 42

بعد إصلاح خلل ما، يعيد messenger:failed:retry تشغيل الرسائل العالقة. وإذا كان الخطأ نهائياً (مستخدم محذوف، بيانات غير صالحة)، فارفع الاستثناء UnrecoverableMessageHandlingException داخل المعالِج: فلن تُعاد محاولة الرسالة دون جدوى.

أخطاء شائعة

• معالِجات غير متساوية الأثر (non-idempotent): يضمن Messenger التسليم «مرة واحدة على الأقل». قد تُعالَج رسالة مرتين بعد انهيار ما، ويجب أن يتحمّل المعالِج ذلك دون إرسال دفعتين أو رسالتين إلكترونيتين.
• انقطاع الاتصال بقاعدة البيانات: قد يغلق خادم MySQL اتصال عامل بقي خاملاً مدة طويلة. يتحقق الوسيط (middleware) doctrine_ping_connection من الاتصال قبل كل رسالة.
• رسائل ثقيلة: لا تضع فيها كيانات ولا ملفات ولا خدمات. المعرّفات والقيم البسيطة تكفي.
• الاختبارات: في بيئة الاختبار، وجّه الرسائل إلى وسيلة النقل in-memory:// للتحقق من أن رسالة ما قد أُرسلت، أو إلى sync:// لتنفيذها فوراً.

مراقبة العمّال

• استخدم systemd لإدارة العمّال في الإنتاج
• اضبط --time-limit لتجنّب تسرّب الذاكرة
• راقب الطابور failed لرصد الرسائل الفاشلة
• استخدم وسيط التسجيل (logging middleware) لتصحيح الأخطاء
• أوقف العمّال بشكل نظيف عند كل نشر باستخدام messenger:stop-workers
• تابع حجم الطوابير بالأمر messenger:stats لاكتشاف أي عامل متوقف

متى لا تلجأ إلى المعالجة غير المتزامنة

إذا كان المستخدم يحتاج إلى النتيجة لمتابعة مساره (دفعة يجب التحقق منها، مخزون يجب حجزه)، فيجب أن تبقى المعالجة متزامنة أو أن تُرفق بآلية متابعة، كحالة (status) تستعلم عنها الواجهة الأمامية. كما تضيف المعالجة غير المتزامنة بنية تحتية تحتاج إلى مراقبة: فلمهمة تستغرق بضع مللي ثوانٍ، لا فائدة منها. خصّصها للمهام البطيئة أو الهشّة أو غير الأساسية للاستجابة.

باختصار: رسائل صغيرة وغير قابلة للتعديل، ومعالِجات متساوية الأثر، ووسيلة نقل للإخفاقات مُعدّة، وعمّال يُعاد تشغيلهم بانتظام ويُوقَفون بشكل نظيف عند كل نشر. بهذه القواعد القليلة، يصبح Messenger أساساً موثوقاً لامتصاص ذروات الحمل والحفاظ على أزمنة استجابة قصيرة.