PHP Academy · درس

العمل مع RabbitMQ في PHP

انشر الرسائل واستهلكها باستخدام RabbitMQ.

العمل مع RabbitMQ في PHP درس مجاني في PHP Academy على CoddyKit. هذا هو الدرس 2 من أصل 4. يمكنك قراءة الدرس كاملاً أدناه مجاناً — ثم تمرن عليه مباشرة في المتصفح باستخدام محرر أكواد مدمج ومدرس ذكاء اصطناعي متاح 24/7. هذا الدرس جزء من مسار التعلم في PHP Academy، وتقدمك يتزامن عبر الويب وتطبيق CoddyKit. تتضمن دورة PHP Academy 4 دروس في المجموع.

RabbitMQ في PHP

RabbitMQ هو وسيط يتحدث بروتوكول AMQP 0-9-1. والعميل القياسي في PHP هو php-amqplib/php-amqplib (مكتبة PHP خالصة) أو ext-amqp المدعوم بـ C. يستخدم هذا الدرس php-amqplib لأنه يعمل في أي بيئة يعمل فيها Composer.

يتضمن نموذج AMQP الأساسي ثلاثة مكونات: تستقبل التبادلات الرسائل، وتوجّهها الروابط حسب المفتاح، وتحتفظ بها الطوابير للمستهلكين. أتقن هذا النموذج، وما تبقى مجرد تفاصيل.

تثبيت العميل

أضف المكتبة باستخدام Composer. فهي تحتاج إلى الامتدادين sockets وbcmath، وهما شائعان في إصدارات PHP المخصصة لسطر الأوامر.

composer require php-amqplib/php-amqplib
# Connection target, e.g. amqp://guest:guest@localhost:5672/

نموذج التبادل/الطابور/الربط

ينشر المنتجون الرسائل إلى تبادل، وليس إلى طابور مباشرة. ويحدد نوع التبادل أسلوب التوجيه:

direct — تطابق تام مع مفتاح التوجيه.
topic — أنماط أحرف بدل مثل order.*.eu.
fanout — بث إلى جميع الطوابير المرتبطة.
headers — مطابقة سمات الترويسة.

يربط الربط طابورًا بتبادل باستخدام نمط توجيه. وفصل المنتجين عن بنية الطوابير هو الغاية الأساسية من طبقة التبادل.

الاتصال والتعريف

افتح اتصالًا، واحصل على قناة، وعرّف طوبولوجيتك. يضمن durable: true بقاء التبادل وقائمة الانتظار بعد إعادة تشغيل وسيط الرسائل. لا تتأثر عمليات التعريف بالتكرار — إذ إن تعريف كيان موجود بالمعطيات المطابقة لا ينفذ أي إجراء.

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();

$ch->exchange_declare('orders', 'topic', false, true, false);
$ch->queue_declare('orders.email', false, true, false, false);
$ch->queue_bind('orders.email', 'orders', 'order.created');

echo "Topology ready\n";
$ch->close();
$conn->close();

نشر رسالة

غلّف المحتوى داخل AMQPMessage. اضبط delivery_mode = 2 لجعل الرسالة مستمرة — إذ إن اجتماع قائمة انتظار دائمة ورسالة مستمرة هو ما يضمن بقاءهما بعد إعادة التشغيل، ولا يكفي أحدهما دون الآخر.

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();

$payload = json_encode(['orderId' => 42, 'total' => 19.90]);
$msg = new AMQPMessage($payload, [
    'content_type'  => 'application/json',
    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
    'message_id'    => bin2hex(random_bytes(8)),
]);

$ch->basic_publish($msg, 'orders', 'order.created');
echo "Published\n";
$ch->close();
$conn->close();

الاستهلاك مع الإقرار اليدوي

سجّل دالة رد اتصال باستخدام basic_consume. مرّر no_ack = false حتى تتحكم في الإقرار. أرسل الإقرار فقط بعد نجاح العمل؛ وعند الفشل، يتيح basic_nack مع إعادة الإدراج لـ RabbitMQ إعادة تسليم الرسالة.

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();
$ch->basic_qos(null, 10, null); // prefetch 10

$callback = function ($msg) {
    $data = json_decode($msg->getBody(), true);
    try {
        // ... do work ...
        $msg->ack();
    } catch (\Throwable $e) {
        $msg->nack(true); // requeue
    }
};

$ch->basic_consume('orders.email', '', false, false, false, false, $callback);
while ($ch->is_consuming()) {
    $ch->wait();
}

الجلب المسبق والتوزيع العادل

بشكل افتراضي، يوزّع RabbitMQ الرسائل بالتناوب على المستهلكين من دون مراعاة مدى انشغال كل منهم — فيتراكم العمل لدى المستهلك البطيء. تحدد basic_qos(null, prefetch, null) الحد الأقصى لعدد الرسائل غير المُقَرّ بها التي يمكن للمستهلك الاحتفاظ بها.

اضبط قيمة الجلب المسبق على رقم صغير، مثل 1–10، للمهام الثقيلة وغير المتساوية حتى يرسل وسيط الرسائل عملًا جديدًا فقط إلى المستهلكين الذين لديهم سعة متاحة. يُسمى هذا التوزيع العادل.

تأكيدات الناشر

تعود basic_publish فورًا ولا تخبرك بما إذا كان وسيط الرسائل قد قبل الرسالة. لضمان النشر، فعّل تأكيدات الناشر: يرسل وسيط الرسائل إقرارًا بمجرد حفظ الرسالة أو توجيهها بأمان.

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();
$ch->confirm_select(); // enable confirms on this channel

$ch->set_ack_handler(fn($m) => print("confirmed\n"));
$ch->set_nack_handler(fn($m) => print("REJECTED\n"));

$ch->basic_publish(new AMQPMessage('hi'), 'orders', 'order.created');
$ch->wait_for_pending_acks(5.0); // block until confirmed or timeout

الرسائل الميتة

اضبط الوسيط x-dead-letter-exchange لقائمة الانتظار حتى تُوجَّه الرسائل المرفوضة، أي nack من دون إعادة إدراج، أو الرسائل المنتهية الصلاحية إلى DLX. ويمكنك دمج ذلك مع x-delivery-limit في قوائم انتظار quorum للحد من عمليات إعادة المحاولة تلقائيًا.

<?php
use PhpAmqpLib\Wire\AMQPTable;

$args = new AMQPTable([
    'x-dead-letter-exchange'    => 'orders.dlx',
    'x-dead-letter-routing-key' => 'order.failed',
    'x-message-ttl'             => 60000, // ms before expiry
]);
// false=passive, true=durable, false=exclusive, false=autodelete, args
$ch->queue_declare('orders.email', false, true, false, false, false, $args);
$ch->queue_declare('orders.dead', false, true, false, false);
$ch->queue_bind('orders.dead', 'orders.dlx', 'order.failed');

تشغيل العمال في بيئة الإنتاج

إليك بعض القواعد التشغيلية المستخلصة من الخبرة لتشغيل عمال PHP RabbitMQ:

شغّل المستهلكين كعمليات CLI طويلة التشغيل تحت إشراف systemd أو Supervisor، بحيث يعيد تشغيلها عند خروجها.
تسرّب PHP الذاكرة مع مرور الوقت — أعد تشغيل العامل بعد معالجة N رسالة أو عند تجاوز حد معيّن للذاكرة.
أرسل heartbeats الخاصة بـ AMQP وتعامل مع SIGTERM للإيقاف السلس، أي أكمل معالجة الرسالة الجارية ثم توقّف.
استخدم قوائم انتظار quorum للتوافر العالي بدلًا من قوائم الانتظار المنسوخة القديمة.

الإيقاف السلس

ترسل عمليات النشر SIGTERM. ويموت العامل الساذج أثناء معالجة الرسالة، مما يفرض إعادة تسليمها. ثبّت معالجًا للإشارة يغيّر قيمة راية؛ ثم أكمل معالجة الرسالة الحالية، وأرسل إقرارها، واخرج من حلقة الاستهلاك بصورة نظيفة. تتيح pcntl_async_signals(true) لـ PHP تسليم الإشارة بين فترات انتظار AMQP.

<?php
pcntl_async_signals(true);
$running = true;
pcntl_signal(SIGTERM, function () use (&$running) {
    $running = false; // stop after the current message
    echo "SIGTERM: draining...\n";
});

while ($running && $ch->is_consuming()) {
    try {
        $ch->wait(null, false, 5); // wakes for signals
    } catch (\PhpAmqpLib\Exception\AMQPTimeoutException $e) {
        // idle tick - loop and re-check $running
    }
}
$ch->close();
echo "Stopped cleanly\n";

تحقق سريع

البقاء بعد إعادة تشغيل وسيط الرسائل.

مراجعة

يمكنكم الآن بناء خط أنابيب حقيقي باستخدام RabbitMQ في PHP:

انشر إلى تبادل، ووجّه عبر عمليات الربط إلى قوائم الانتظار.
يضمن اجتماع قائمة انتظار دائمة + رسالة مستمرة البقاء بعد إعادة التشغيل.
استهلك مع إقرار يدوي واضبط basic_qos prefetch لتحقيق التوزيع العادل.
استخدم تأكيدات الناشر لضمان عمليات الإرسال وDLX للتعامل مع الرسائل السامة.
شغّل العمال تحت إشراف supervisor مع heartbeats وإيقاف سلس.

التالي: Kafka، عندما تحتاجون إلى بث عالي الإنتاجية بدلًا من وضع المهام في قوائم انتظار.

الأسئلة الشائعة

هل درس «العمل مع RabbitMQ في PHP» مجاني؟

نعم — نص درس «العمل مع RabbitMQ في PHP» كامل متاح مجاناً هنا على الويب. لتمرينه بشكل تفاعلي (محرر أكواد مدمج ومدرس ذكاء اصطناعي متاح 24/7) وفتح باقي دورة PHP Academy، انتقل إلى CoddyKit PRO. تتضمن دورة PHP Academy 4 دروس في المجموع.

ماذا ستتعلم في «العمل مع RabbitMQ في PHP»؟

انشر الرسائل واستهلكها باستخدام RabbitMQ. تتمرن على PHP Academy مع أكواد عملية تشغلها مباشرة في المتصفح، ومدرس ذكاء اصطناعي متاح 24/7 يجيب على أسئلتك أثناء عملك.

هل أحتاج إلى خبرة سابقة لأبدأ PHP Academy؟

لا تُشترط خبرة سابقة. PHP Academy على CoddyKit منظم للمبتدئين حتى المتقدمين، لذا يمكنك البدء من هنا أو من البداية والتقدم بسرعتك الخاصة. هذا هو الدرس 2 من أصل 4.

كم من الوقت يستغرق درس «العمل مع RabbitMQ في PHP»؟

معظم دروس CoddyKit تستغرق حوالي 5–10 دقائق. كل منها موجز وتفاعلي، لذا تحرز تقدماً مستمراً وتستأنف من حيث توقفت عبر الويب والتطبيق.

هل يمكنني كتابة وتشغيل أكواد في درس PHP Academy هذا؟

نعم. كل درس في PHP Academy يتضمن محرر أكواد مدمج، لذا تكتب وتشغل أكواداً حقيقية مباشرة في متصفحك وتحصل على تعليقات فورية من الذكاء الاصطناعي — بدون إعداد محلي.

جميع الدروس في هذه الدورة

← العودة إلى PHP Academy