RabbitMQ: طوابير موثوقة لتطبيقاتك على سيرفرك

في مقال مفاهيم الطوابير شغلنا RabbitMQ على سيرفر واحد، ورأينا التأكيد Ack والتوزيع بين ال Workers وإعادة المحاولة وال Dead Letter Queue، وفي مقال الطابور أم بث الأحداث عرفنا متى يكون الطابور هو الأداة الصحيحة. لكن الوسيط الواحد نقطة فشل واحدة، فإذا توقف السيرفر توقفت معه كل الطوابير، وإذا تعطل القرص ضاعت الرسائل التي لم تعالج بعد، لذلك يأتي هذا المقال ليأخذ RabbitMQ خطوة نحو الإنتاج: عنقود Cluster من ثلاث عقد، وتوجيه الرسائل بالموزعات Exchanges، وطوابير Quorum تبقى بعد سقوط عقدة، ونختبر ذلك بإيقاف عقدة بالقوة أثناء النشر ونعد الرسائل واحدة واحدة، وسوف نناقش في هذا المقال ما يلي:

• ما هو RabbitMQ، ومتى يناسبك ومتى لا يناسبك.
• الرخصة MPL 2.0، وعلاقة Broadcom بالمشروع، وما يعنيه ذلك لمن يشغله بنفسه.
• الموزعات Exchanges من نوع Direct وTopic وFanout، والروابط Bindings، وماذا يحدث للرسالة التي لا تجد طابوراً.
• أنواع الطوابير: Classic وQuorum وStream، ولماذا Quorum هو الاختيار الافتراضي للعمل المهم.
• تأكيد النشر Publisher Confirms مع العنقود، وال Prefetch، ومدة الصلاحية TTL مع ال Dead Letter Exchange.
• تجربة عملية على عنقود من ثلاث عقد: نوقف العقدة القائدة أثناء النشر مع التأكيد ودونه، ونرى أثر ال Prefetch على عدالة التوزيع.
• واجهة الإدارة، والنسخ الاحتياطي والمراقبة والحماية، وأخطاء شائعة.

ما هو RabbitMQ، ومتى يناسبك؟

RabbitMQ وسيط رسائل Message Broker مفتوح المصدر مكتوب بلغة Erlang، وهو من أقدم الوسطاء وأكثرها انتشاراً، ويتكلم بحسب صفحة البروتوكولات بروتوكولي AMQP 0-9-1 وAMQP 1.0 بشكل أساسي، ويدعم MQTT وSTOMP عبر إضافات Plugins، لذلك تجد له مكتبات في كل لغة تقريباً. وفكرته الأساسية أن المنتج لا يرسل إلى الطابور مباشرة، وإنما إلى موزع Exchange يقرر بالقواعد التي تضعها أي طابور يأخذ الرسالة، ثم يأخذ ال Worker الرسالة من الطابور ويؤكدها بعد أن ينتهي منها، وهذا هو الشكل الذي شرحناه في مقال المفاهيم.

ويناسبك RabbitMQ في الحالات التالية:

• المهام الخلفية Jobs: إرسال البريد، وإنشاء الفواتير، وتصغير الصور، وكل عمل يحتاج تأكيداً لكل رسالة وإعادة محاولة وDead Letter Queue، وهذه الأشياء جاهزة فيه دون أن تبنيها بنفسك.
• التوجيه المرن Routing: رسالة واحدة تذهب إلى طوابير مختلفة بحسب نوعها أو الدولة أو الأولوية، دون أن يعرف المنتج من يستهلكها.
• أنظمة بلغات مختلفة: خدمة بلغة Java تنشر وخدمة بلغة Python تستهلك، والبروتوكول نفسه في الطرفين.

ولا يناسبك في الحالات التالية:

• سجل أحداث يعاد قراءته: إذا كان أكثر من نظام يريد قراءة الأحداث نفسها بعد أيام، أو تحتاج إعادة المعالجة من نقطة قديمة، فهذه طبيعة سجل الأحداث مثل Kafka، كما شرحنا في مقال الطابور أم بث الأحداث، وإن كان RabbitMQ يقدم Streams كما سيأتي.
• حجم صغير وقاعدة بيانات موجودة: آلاف المهام في اليوم يكفيها طابور فوق جدول في PostgreSQL، دون سيرفر جديد تراقبه.
• رسائل بين الخدمات لا تحتاج تخزيناً: طلب ورد سريع بين خدمات مصغرة، أو أخبار لحظية لا قيمة لها بعد ثوان، وهنا يكون وسيط خفيف مثل NATS أنسب.

الرخصة، وعلاقة Broadcom بالمشروع

الكود الأساسي ل RabbitMQ وإضافاته الأساسية برخصة Mozilla Public License 2.0 كما يقول ملف الرخصة في المستودع، وهي رخصة مفتوحة المصدر تسمح لك بتشغيله في عملك وتقديمه ضمن خدمة تبيعها دون أي قيد، والشرط الوحيد المهم أنك إذا عدلت ملفات الكود نفسها ووزعتها فعليك أن تنشر تعديلاتك على تلك الملفات بالرخصة نفسها. وهذا يختلف عن أدوات غيرت رخصتها إلى رخص غير مفتوحة، مثل Redpanda التي سوف نتكلم عنها في مقال قادم من المرحلة.

أما الملكية فالمشروع انتقل بين عدة شركات حتى وصل إلى VMware، ثم أصبح تحت Broadcom عندما أكملت Broadcom شراء VMware في نوفمبر 2023، وتجد اسم Broadcom اليوم في أسفل الموقع الرسمي مع عرض لنسخة تجارية بدعم فني. والذي يهمك عملياً من هذا هو سياسة الدعم، فبحسب صفحة الإصدارات ينتهي دعم المجتمع Community Support لكل سلسلة إصدارات بعد فترة قصيرة من صدور السلسلة التالية، بينما يستمر الدعم التجاري سنوات، فمثلاً انتهى دعم المجتمع لسلسلة 4.2 في 31 يوليو 2026، ويستمر دعمها التجاري حتى 30 يونيو 2030. وهذا يعني أنك إذا كنت تشغل النسخة المجانية فعليك أن تبقى على آخر سلسلة وتخطط للترقية كل بضعة أشهر، والسبب أن إصلاحات الأمان للسلاسل القديمة لم تعد تصل مجاناً.

الموزعات Exchanges والروابط Bindings

الموزع Exchange هو المكان الذي ينشر إليه المنتج، والرابط Binding قاعدة تقول «أرسل إلى هذا الطابور الرسائل التي مفتاح توجيهها Routing Key يطابق كذا»، ويشرح توثيق الموزعات الأنواع التي تحتاجها في أغلب الأنظمة:

• Direct: يرسل الرسالة إلى الطوابير التي مفتاح ربطها يساوي مفتاح التوجيه تماماً، مثل paid.
• Topic: يطابق مفتاح التوجيه بنمط، والمفتاح كلمات تفصلها نقاط مثل order.created.sa، والرمز * يعني كلمة واحدة بالضبط، والرمز # يعني صفراً أو أكثر من الكلمات.
• Fanout: يرسل نسخة إلى كل طابور مربوط به ويتجاهل المفتاح، وهو النشر والاشتراك Publish-Subscribe الذي بنيناه في تجربة أنماط المعمارية.
• Default Exchange: موزع من نوع Direct موجود دائماً دون اسم، وكل طابور مربوط به تلقائياً باسمه، لذلك عندما تنشر إلى الاسم الفارغ مع مفتاح emails تصل الرسالة إلى طابور emails، وهذا ما استخدمناه في مقال المفاهيم.

والصورة التالية تبين الموزعات الثلاثة كما بنيناها في التجربة لاحقاً:

وقد يتساءل البعض: ماذا يحدث لرسالة لا يطابق مفتاحها أي رابط؟ والإجابة أن الوسيط يحذفها بصمت في الوضع الافتراضي، وهذا خطأ يصعب اكتشافه لأن المنتج لا يرى أي خطأ، والحل أحد أمرين: أن تنشر بالخيار mandatory فيعيد الوسيط الرسالة إلى المنتج، وهو ما سوف تراه في التجربة، أو أن تضع للموزع موزعاً بديلاً Alternate Exchange يستقبل كل رسالة لم تجد طريقها، وتربط به طابوراً تراقبه.

أنواع الطوابير: Classic وQuorum وStream

عندما تعرف طابوراً في RabbitMQ 4 تختار نوعه بالخيار x-queue-type، والفرق بين الأنواع هو أين تعيش الرسالة وكم نسخة منها:

• Classic: الطابور الأقدم، ويعيش على عقدة واحدة فقط، فإذا توقفت العقدة توقف الطابور حتى تعود، وإذا تلف قرصها ضاعت رسائله. ويقول توثيق Quorum Queues إن نسخ الطوابير الكلاسيكية Classic Queue Mirroring أزيل بدءاً من RabbitMQ 4.0، وإن الشكلين الوحيدين المنسوخين على أكثر من عقدة هما Quorum وStream، فإذا وجدت دليلاً قديماً يشرح سياسة ha-mode فهو لا ينطبق على الإصدارات الحالية.
• Quorum: طابور له نسخ على عدة عقد، واحدة منها قائد Leader والبقية تابعون Followers، ويستخدم خوارزمية Raft للإجماع التي شرحناها في مقال الأنظمة الموزعة، وعدد النسخ الافتراضي ثلاث، ولا يعطي الوسيط المنتج تأكيد النشر إلا بعد أن تحفظ أغلبية النسخ الرسالة. ويدعم بحسب التوثيق حد التسليم ومعالجة الرسائل التي تفشل دائماً Poison Messages والانتقال إلى ال Dead Letter Exchange بضمان مرة على الأقل.
• Stream: سجل يضاف إليه فقط ولا تحذف الرسالة منه بعد قراءتها، وهو أقرب إلى Kafka داخل RabbitMQ، وسوف نراه في التجربة باختصار.

والصورة التالية تبين ما يحدث في طابور Quorum على ثلاث عقد عندما تسقط العقدة القائدة، وهو ما سوف نختبره بعد قليل:

وهناك ثلاث قواعد يذكرها توثيق Quorum Queues يجب أن تعرفها قبل أن تبني العنقود:

• عدد العقد فردي: ثلاث أو خمس، والسبب أن الأغلبية من أربع عقد هي ثلاث، فالعقدة الرابعة لا تزيد عدد الأعطال التي يتحملها الطابور، وأي حجم أكبر من خمس نسخ يبطئ الطابور بشكل ملحوظ.
• فقدان الأغلبية يوقف الطابور: إذا توقفت عقدتان من ثلاث توقف الطابور عن قبول الرسائل حتى تعود إحداهما، وإذا ضاعت عقدتان نهائياً فالتوثيق يقول إن الطابور يصبح غير متاح بشكل دائم ويجب حذفه بالقوة وإعادة إنشائه، وسوف نكسر العنقود بهذه الطريقة في التجربة.
• الذاكرة تكبر مع طول الطابور: بحسب التوثيق يستهلك طابور Quorum واحد على الأقل 1 ميجابايت لكل 30 ألف رسالة فيه، لذلك الطابور الذي يكبر بالملايين لأن ال Workers متوقفون يضغط على ذاكرة العقد كلها.

ويجدر الإشارة هنا إلى أن العنقود نفسه، أي قائمة الطوابير والمستخدمين والموزعات، أصبح يحفظ منذ RabbitMQ 4.2 في مخزن اسمه Khepri يعتمد على Raft أيضاً، كما يذكر توثيق مخزن البيانات الوصفية، وهذا سبب إضافي لأن يكون عدد العقد فردياً.

تأكيد النشر وال Prefetch في العنقود

شرحنا تأكيد النشر Publisher Confirms وال Prefetch في مقال المفاهيم، وما يضيفه العنقود هو معنى التأكيد: في طابور Quorum لا يصل التأكيد إلا بعد أن تحفظ أغلبية العقد الرسالة، فإذا وصلك التأكيد فالرسالة تبقى ولو سقطت العقدة التي اتصلت بها بعد لحظة، وإذا لم يصلك فعليك أن تفترض أنها لم تصل وتعيد إرسالها. وهذا يعني شيئين عمليين:

• لا تنتظر التأكيد لكل رسالة وحدها: عندما كتبنا الناشر أول مرة بمكتبة pika ينتظر تأكيد كل رسالة قبل أن يرسل التالية كانت السرعة قرابة 20 رسالة في الثانية فقط، والسبب أن كل رسالة تنتظر النسخ إلى عقدة أخرى والكتابة على القرص قبل أن تبدأ التالية، وعندما أرسلنا دفعات من 500 رسالة وانتظرنا تأكيداتها معاً ارتفعت السرعة إلى قرابة 650 رسالة في الثانية على الجهاز نفسه. ويصف توثيق الناشرين انتظار التأكيد لكل رسالة وحدها بأنه غير منصوح به إطلاقاً لأثره الكبير على السرعة، وينصح بالتأكيدات غير المتزامنة أو بالدفعات.
• إعادة الإرسال تنتج تكراراً: إذا وصلت الرسالة وضاع تأكيدها مع سقوط العقدة فسوف ترسلها مرة ثانية، لذلك يجب أن يكون المستهلك آمناً عند التكرار كما شرحنا في مقال Idempotency، وسوف نرى هذا التكرار بالأرقام في التجربة.

مدة الصلاحية TTL وال Dead Letter Exchange

رأينا في مقال المفاهيم كيف تنتقل الرسالة إلى ال Dead Letter Queue بعد حد المحاولات، وهناك سبب آخر مهم لانتقالها هو انتهاء صلاحيتها، فبحسب توثيق TTL تستطيع أن تحدد مدة قصوى لبقاء الرسالة في الطابور بالخيار x-message-ttl، وعندما تنتهي المدة تحذف الرسالة أو تنتقل إلى ال Dead Letter Exchange إذا عرفته. والاستخدام الشائع لهذا هو العمل الذي يفقد قيمته بعد وقت، مثل طلب دفع لم يعالج خلال دقيقة فتنقله إلى طابور مراجعة بدل أن يعالج بعد ساعة وقد ألغى العميل الطلب، وسوف نرى ذلك في التجربة بمدة ثلاث ثوان.

تجربة عملية: عنقود من ثلاث عقد

الآن سوف نبني عنقوداً من ثلاث عقد RabbitMQ 4.3.6 في Docker، ونشغل عليه كل ما سبق بمكتبة pika 1.4.4 ومكتبة aio-pika 10.1.1 في Python 3.13، ونفذنا الأوامر من Git Bash على Windows مع Docker Engine 29.8.0، وهي تعمل كما هي على Linux. ولا يوجد منفذ منشور على الجهاز إلا واجهة الإدارة، وهي مربوطة بالعنوان 127.0.0.1 فقط.

ملفات التجربة

أنشئ مجلداً فيه الملفات التالية، أولها rabbitmq.conf الذي يخبر كل عقدة بأسماء العقد الأخرى:

# the three nodes find each other from this fixed list
cluster_formation.peer_discovery_backend = classic_config
cluster_formation.classic_config.nodes.1 = rabbit@rmq1
cluster_formation.classic_config.nodes.2 = rabbit@rmq2
cluster_formation.classic_config.nodes.3 = rabbit@rmq3
default_user = lab
default_pass = lab-pass-2026

ثم compose.yaml:

name: lab-rabbitmq

x-node: &node
  image: rabbitmq:4.3.6-management-alpine
  environment:
    # every node and the CLI tools must share the same Erlang cookie
    RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS: "-setcookie lab-cookie-2026"
    RABBITMQ_CTL_ERL_ARGS: "-setcookie lab-cookie-2026"
  healthcheck:
    test: ["CMD", "rabbitmq-diagnostics", "-q", "check_port_connectivity"]
    interval: 5s
    retries: 40

services:
  rmq1:
    <<: *node
    hostname: rmq1
    container_name: lab-rabbitmq-rmq1
    volumes: ["./rabbitmq.conf:/etc/rabbitmq/conf.d/90-lab.conf:ro", "rmq1:/var/lib/rabbitmq"]
    ports: ["127.0.0.1:18672:15672"]   # management UI, from this machine only
  rmq2:
    <<: *node
    hostname: rmq2
    container_name: lab-rabbitmq-rmq2
    volumes: ["./rabbitmq.conf:/etc/rabbitmq/conf.d/90-lab.conf:ro", "rmq2:/var/lib/rabbitmq"]
  rmq3:
    <<: *node
    hostname: rmq3
    container_name: lab-rabbitmq-rmq3
    volumes: ["./rabbitmq.conf:/etc/rabbitmq/conf.d/90-lab.conf:ro", "rmq3:/var/lib/rabbitmq"]

  app:
    build: .
    image: lab-rabbitmq-app:1
    environment:
      NODES: "rmq2,rmq1,rmq3"
      AMQP_USER: lab
      AMQP_PASS: lab-pass-2026
    profiles: ["tools"]

volumes:
  rmq1:
  rmq2:
  rmq3:

في الإعداد أعلاه لاحظ التالي:

• العقد تتعارف بقائمة ثابتة classic_config، وكل عقدة اسمها rabbit@ متبوعاً باسم المضيف Hostname، لذلك يجب أن يكون ال Hostname ثابتاً، والسبب أن البيانات على القرص مرتبطة باسم العقدة.
• كل العقد تشترك في كلمة سر Erlang Cookie واحدة، وهي التي تسمح لها بالاتصال ببعضها، ولأدوات سطر الأوامر مثل rabbitmqctl أيضاً، وفي الإنتاج اجعلها قيمة طويلة عشوائية واحفظها كما تحفظ كلمات المرور.
• ال Image من نوع management، أي أن واجهة الإدارة مفعلة فيها، وهي على المنفذ 15672 داخل ال Container ونربطها بالمنفذ 18672 على الجهاز.
• الخدمة app هي ال Container الذي نشغل منه سكربتات Python، وقائمة NODES تبدأ ب rmq2 لأننا سوف نوقف هذه العقدة تحديداً.

ثم Dockerfile للسكربتات:

FROM python:3.13.16-slim
RUN pip install --no-cache-dir pika==1.4.4 aio-pika==10.1.1
WORKDIR /app
COPY *.py ./
ENV PYTHONUNBUFFERED=1

ثم common.py، وفيه دالة الاتصال التي تجرب العقد واحدة بعد الأخرى:

import os, pika

def connect(skip=()):
    """Connect to the first node in NODES that answers (a client-side failover list)."""
    creds = pika.PlainCredentials(os.environ["AMQP_USER"], os.environ["AMQP_PASS"])
    for host in os.environ["NODES"].split(","):
        if host in skip:
            continue
        try:
            conn = pika.BlockingConnection(pika.ConnectionParameters(
                host=host, credentials=creds, connection_attempts=1, socket_timeout=3))
            return conn, host
        except (pika.exceptions.AMQPConnectionError, OSError):
            continue
    raise SystemExit("no node is reachable")

أما بقية السكربتات فسوف نعرض كل واحد منها في الخطوة التي يستخدم فيها.

تشغيل العنقود

docker compose --profile tools build
docker compose up -d --wait rmq1 rmq2 rmq3
docker exec lab-rabbitmq-rmq1 rabbitmqctl cluster_status

والمخرج بعد حذف بعض الأقسام سوف يكون كما يلي:

Cluster status of node rabbit@rmq1 ...
Basics

Cluster name: rabbit@rmq1
Total CPU cores available cluster-wide: 24

Disk Nodes

rabbit@rmq1
rabbit@rmq2
rabbit@rmq3

Running Nodes

rabbit@rmq1
rabbit@rmq2
rabbit@rmq3

Versions

rabbit@rmq1: RabbitMQ 4.3.6 on Erlang 27.3.4.18
rabbit@rmq2: RabbitMQ 4.3.6 on Erlang 27.3.4.18
rabbit@rmq3: RabbitMQ 4.3.6 on Erlang 27.3.4.18

لاحظ أن العقد الثلاث تعمل في عنقود واحد دون أي أمر join_cluster يدوي، والسبب قائمة العقد في ملف الإعداد، ولاحظ أن كل عقدة ترى ثمانية أنوية لأن العقد الثلاث تعمل على الجهاز نفسه، وفي الإنتاج تكون كل عقدة على سيرفر مستقل، والأفضل في منطقة توفر Availability Zone مختلفة.

التوجيه بالموزعات

السكربت routing.py ينشئ الموزعات والطوابير والروابط التي في صورة الموزعات، ثم ينشر ست رسائل بالخيار mandatory، ويطبع عدد الرسائل في كل طابور:

# Exchanges and bindings: where does each message go?
import pika
from common import connect

conn, _ = connect()
ch = conn.channel()
ch.exchange_declare("orders.topic", "topic", durable=True)
ch.exchange_declare("orders.direct", "direct", durable=True)
ch.exchange_declare("orders.fanout", "fanout", durable=True)

binds = {   # queue -> (exchange, binding key)
    "sa-orders":   ("orders.topic", "order.*.sa"),
    "all-orders":  ("orders.topic", "order.#"),
    "paid-orders": ("orders.direct", "paid"),
    "audit":       ("orders.fanout", ""),
    "analytics":   ("orders.fanout", ""),
}
for q, (ex, key) in binds.items():
    ch.queue_declare(q, durable=True, arguments={"x-queue-type": "quorum"})
    ch.queue_bind(q, ex, routing_key=key)

ch.confirm_delivery()
for ex, key in [("orders.topic", "order.created.sa"), ("orders.topic", "order.created.ae"),
                ("orders.topic", "order.paid.sa.riyadh"), ("orders.direct", "paid"),
                ("orders.direct", "refunded"), ("orders.fanout", "anything")]:
    try:
        ch.basic_publish(ex, key, b"{}", mandatory=True)   # mandatory: tell me if no queue got it
        print(f"routed      {ex:14} {key}")
    except pika.exceptions.UnroutableError:
        print(f"UNROUTABLE  {ex:14} {key}")
print()
for q in binds:
    print(f"{q:12} {ch.queue_declare(q, passive=True).method.message_count}")
conn.close()

docker compose run --rm app python routing.py

والمخرج سوف يكون كما يلي:

routed      orders.topic   order.created.sa
routed      orders.topic   order.created.ae
routed      orders.topic   order.paid.sa.riyadh
routed      orders.direct  paid
UNROUTABLE  orders.direct  refunded
routed      orders.fanout  anything

sa-orders    1
all-orders   3
paid-orders  1
audit        1
analytics    1

في المخرج أعلاه لاحظ التالي:

• order.created.sa وصلت إلى طابورين: sa-orders لأنها تطابق order.*.sa، وall-orders لأنها تطابق order.#، فالموزع ينسخ الرسالة لكل رابط يطابقها.
• order.paid.sa.riyadh لم تصل إلى sa-orders، والسبب أن * تطابق كلمة واحدة فقط، والمفتاح هنا فيه أربع كلمات، وهذا خطأ شائع عندما يضيف أحد الفريق كلمة إلى المفتاح فتتوقف رسائل عن الوصول دون أي خطأ.
• refunded لم تجد أي رابط، ولأننا نشرنا بالخيار mandatory مع تأكيد النشر أعاد الوسيط الرسالة فعرف المنتج بذلك، ولو نشرنا دونه لحذفت بصمت.
• رسالة Fanout وصلت إلى audit وanalytics مع أن مفتاحها anything لا علاقة له بأي رابط.

رسالة تنتهي صلاحيتها فتنتقل إلى طابور آخر

السكربت ttl_dlx.py ينشئ طابوراً مدة بقاء الرسالة فيه ثلاث ثوان، ويربطه بال Default Exchange كموزع للرسائل الميتة، فتنتقل الرسالة المنتهية إلى payments.expired:

# A message that nobody consumes within 3 seconds moves to another queue through a dead letter exchange
import time
from common import connect

conn, _ = connect()
ch = conn.channel()
ch.queue_declare("payments.expired", durable=True, arguments={"x-queue-type": "quorum"})
ch.queue_declare("payments.pending", durable=True, arguments={
    "x-queue-type": "quorum",
    "x-message-ttl": 3000,                           # each message may wait 3 s at most
    "x-dead-letter-exchange": "",
    "x-dead-letter-routing-key": "payments.expired",
})
ch.confirm_delivery()
ch.basic_publish("", "payments.pending", b'{"payment": 77}')
for t in (0, 4):
    time.sleep(t)
    p = ch.queue_declare("payments.pending", passive=True).method.message_count
    e = ch.queue_declare("payments.expired", passive=True).method.message_count
    print(f"after {t}s: pending={p} expired={e}")
m, props, body = ch.basic_get("payments.expired", auto_ack=True)
print("x-first-death-reason:", props.headers["x-first-death-reason"], body.decode())
conn.close()

docker compose run --rm app python ttl_dlx.py

والمخرج سوف يكون كما يلي:

after 0s: pending=1 expired=0
after 4s: pending=0 expired=1
x-first-death-reason: expired {"payment": 77}

لاحظ أن أحداً لم يقرأ الرسالة، ومع ذلك خرجت من payments.pending بعد ثلاث ثوان ووصلت إلى payments.expired كما هي، ومعها الترويسة x-first-death-reason بقيمة expired، بينما كانت القيمة في مقال المفاهيم delivery_limit، فالترويسة نفسها تخبرك لماذا وصلت الرسالة.

Stream: قارئان يقرآن الرسائل نفسها

السكربت stream.py ينشئ Stream اسمه events مدة حفظه سبعة أيام، وينشر ثلاث رسائل، ثم يقرؤها قارئان من البداية:

# A stream keeps messages after they are read: two readers both start from the first message
from common import connect

conn, _ = connect()
ch = conn.channel()
ch.queue_declare("events", durable=True, arguments={"x-queue-type": "stream", "x-max-age": "7D"})
ch.confirm_delivery()
for i in range(1, 4):
    ch.basic_publish("", "events", f"event-{i}".encode())

def read(who):
    got = []
    ch.basic_qos(prefetch_count=10)              # a stream consumer must set a prefetch
    def on(c, m, p, body):
        got.append(body.decode())
        c.basic_ack(m.delivery_tag)
        if len(got) == 3:
            c.stop_consuming()
    ch.basic_consume("events", on, arguments={"x-stream-offset": "first"})
    ch.start_consuming()
    print(f"[{who}] {got}")

read("billing")
read("analytics")
conn.close()

docker compose run --rm app python stream.py

والمخرج سوف يكون كما يلي:

[billing] ['event-1', 'event-2', 'event-3']
[analytics] ['event-1', 'event-2', 'event-3']

لاحظ أن القارئ الثاني وجد الرسائل الثلاث مع أن الأول أكدها، وهذا عكس ما رأيناه في الطابور العادي في مقال الطابور أم بث الأحداث، والخيار x-stream-offset يحدد من أين يبدأ القارئ: first من أول ما هو محفوظ، أو last وnext أو وقت محدد. ويكفيك هذا إذا كنت تحتاج سجلاً صغيراً بجانب طوابيرك، أما إذا كان السجل هو أساس نظامك فانظر Kafka.

نوقف العقدة القائدة أثناء النشر، مع التأكيد

هذه هي الخطوة الأهم، فالسكربت publish.py ينشر عشرين ألف رسالة مرقمة إلى طابور Quorum اسمه jobs على دفعات من 500، وفي وضع confirms ينتظر تأكيد كل دفعة، فإذا انقطع الاتصال اتصل بعقدة أخرى وأعاد الدفعة التي لم يصلها تأكيدها، وفي وضع fire يكتب الرسائل في الاتصال ويكمل دون أن ينتظر أي تأكيد. افتح البطاقة التالية لترى السكربت كاملاً:

# Publishes N numbered messages to the quorum queue "jobs", 500 at a time.
#   python publish.py confirms N  -> waits for the broker's confirms of each batch, resends what was not confirmed
#   python publish.py fire N      -> no confirms: write to the socket and move on
import asyncio, os, sys, time
import logging
import aio_pika

logging.getLogger("aiormq").setLevel(logging.CRITICAL)   # we print our own short error line

MODE, TOTAL, BATCH = sys.argv[1], int(sys.argv[2]), 500
USER, PASS = os.environ["AMQP_USER"], os.environ["AMQP_PASS"]

async def connect(skip=None):
    for host in os.environ["NODES"].split(","):
        if host == skip:
            continue
        try:
            conn = await aio_pika.connect(f"amqp://{USER}:{PASS}@{host}/", timeout=3)
            ch = await conn.channel(publisher_confirms=(MODE == "confirms"))
            await ch.declare_queue("jobs", durable=True, arguments={"x-queue-type": "quorum"})
            return conn, ch, host
        except Exception:
            continue
    raise SystemExit("no node is reachable")

async def main():
    conn, ch, host = await connect()
    print(f"connected to {host}, mode={MODE}")
    start, i, errors = time.time(), 0, 0
    while i < TOTAL:
        batch = range(i, min(i + BATCH, TOTAL))
        try:
            sends = [ch.default_exchange.publish(
                        aio_pika.Message(str(n).encode(), delivery_mode=aio_pika.DeliveryMode.PERSISTENT),
                        routing_key="jobs", timeout=10) for n in batch]
            await asyncio.gather(*sends)    # with confirms: returns when the broker confirmed all of them
            i = batch[-1] + 1
            if i % 5000 == 0:
                print(f"{time.time() - start:5.1f}s sent={i}")
        except Exception as e:
            errors += 1
            print(f"{time.time() - start:5.1f}s batch from {i}: {type(e).__name__}, reconnecting")
            await asyncio.sleep(1)
            if MODE == "fire":
                i = batch[-1] + 1           # without confirms we do not know what arrived, so we move on
            conn, ch, host = await connect(skip=host)
            print(f"{time.time() - start:5.1f}s connected to {host}")
    print(f"done: sent={TOTAL} errors={errors} in {time.time() - start:.1f}s")
    await conn.close()

asyncio.run(main())

ثم count.py الذي يقرأ كل الرسائل من الطابور ويعد الأرقام التي وصلت، والمكررة، والضائعة:

# Drains "jobs" and checks which of the numbered messages arrived
import sys
from common import connect

total = int(sys.argv[1])
conn, _ = connect()
ch = conn.channel()
ch.basic_qos(prefetch_count=1000)
seen, dup = set(), 0
for m, props, body in ch.consume("jobs", auto_ack=True, inactivity_timeout=3):
    if m is None:                      # nothing new for 3 seconds: the queue is empty
        break
    n = int(body)
    dup += n in seen
    seen.add(n)
missing = [n for n in range(total) if n not in seen]
print(f"received={len(seen) + dup} unique={len(seen)} duplicates={dup} lost={len(missing)}")
if missing:
    print(f"lost ids: {len(missing)} between {missing[0]} and {missing[-1]}")
conn.close()

الآن نشغل الناشر في الخلفية، وننتظر حتى ينشر خمسة آلاف رسالة، ثم نوقف العقدة rmq2 بالقوة بالأمر docker kill، وهي العقدة التي اتصل بها الناشر والتي تقود الطابور:

docker compose run --rm -T app python publish.py confirms 20000 > confirms.log 2>&1 &
until grep -q "sent=5000" confirms.log; do sleep 0.2; done
docker kill lab-rabbitmq-rmq2
until grep -q "done:" confirms.log; do sleep 1; done
cat confirms.log
docker exec lab-rabbitmq-rmq1 rabbitmqctl -q list_queues name leader messages
docker compose run --rm -T app python count.py 20000

والمخرج بعد حذف أسطر Compose سوف يكون كما يلي:

connected to rmq2, mode=confirms
 10.9s sent=5000
 13.3s batch from 6000: AMQPConnectionError, reconnecting
 14.4s connected to rmq1
 22.8s sent=10000
 30.0s sent=15000
 35.4s sent=20000
done: sent=20000 errors=1 in 35.4s
jobs	rabbit@rmq3	20272
received=20272 unique=20000 duplicates=272 lost=0

في المخرج أعلاه لاحظ التالي:

• الطابور أنشئ عبر الاتصال بالعقدة rmq2 فأصبحت هي قائده، وقد تحققنا من ذلك بالأمر list_queues أثناء النشر، وبعد التجربة أصبح القائد rmq3، أي أن العقدتين الباقيتين انتخبتا قائداً جديداً دون أي تدخل منا، وأكمل الناشر بعد قرابة ثانية من الانقطاع.
• الناشر فقد الاتصال أثناء الدفعة التي تبدأ بالرسالة 6000، فاتصل بالعقدة rmq1 وأعاد الدفعة كاملة، لأنه لا يعرف أي رسائلها وصلت.
• النتيجة صفر رسائل ضائعة من عشرين ألفاً، و272 رسالة مكررة، وهي رسائل من تلك الدفعة كانت قد حفظت فعلاً لكن تأكيدها ضاع مع العقدة، وهذا هو ضمان «مرة على الأقل»: لا يضيع شيء، لكن قد يصل الشيء مرتين.

التجربة نفسها دون تأكيد

نعيد تشغيل العقدة rmq2، ونحذف الطابور، ونكرر التجربة بالوضع fire:

docker compose up -d --wait rmq2
docker exec lab-rabbitmq-rmq1 rabbitmqctl -q delete_queue jobs
docker compose run --rm -T app python publish.py fire 20000 > fire.log 2>&1 &
until grep -q "sent=5000" fire.log; do sleep 0.1; done
docker kill lab-rabbitmq-rmq2
until grep -q "done:" fire.log; do sleep 1; done
cat fire.log
docker compose run --rm -T app python count.py 20000

والمخرج سوف يكون كما يلي:

connected to rmq2, mode=fire
  2.5s sent=5000
  3.2s batch from 6000: AMQPConnectionError, reconnecting
  4.3s connected to rmq1
  6.3s sent=10000
  9.0s sent=15000
 11.2s sent=20000
done: sent=20000 errors=1 in 11.2s
received=19500 unique=19500 duplicates=0 lost=500
lost ids: 500 between 6000 and 6499

لاحظ أن الناشر أنهى عمله أسرع بثلاث مرات، وطبع sent=20000، لكن 500 رسالة لم تصل أبداً، وهي الدفعة التي كانت في الطريق عندما سقطت العقدة. والسبب أن الناشر دون تأكيد يعرف فقط أنه كتب الرسائل في الاتصال، ولا يعرف هل استلمتها العقدة وحفظتها أم لا، فعندما انقطع الاتصال لم يكن أمامه إلا أن يكمل ويخسر الدفعة، أو يعيدها كلها دون أي معلومة. وعدد الرسائل الضائعة يتغير من تجربة لأخرى بحسب لحظة السقوط، لكن المهم أن الرقم ليس صفراً، وأن الناشر لم يعرف بذلك إلا لأننا عددنا الرسائل بأنفسنا، وفي نظام حقيقي سوف تعرف عندما يسأل العميل عن طلب لم يصل.

نكسر العنقود: عقدتان من ثلاث

الآن نوقف عقدتين معاً ونحاول النشر مع التأكيد إلى العقدة الباقية:

docker kill lab-rabbitmq-rmq2 lab-rabbitmq-rmq3
docker compose run --rm -T -e NODES=rmq1 app python publish.py confirms 500

والمخرج سوف يكون كما يلي:

connected to rmq1, mode=confirms
 10.1s batch from 0: TimeoutError, reconnecting
no node is reachable

لاحظ أن الاتصال بالعقدة rmq1 نجح، لكن التأكيد لم يصل خلال عشر ثوان، والسبب أن عقدة واحدة من ثلاث ليست أغلبية، فلا تستطيع أن تحفظ رسالة باسم الطابور، وهذا هو السلوك الصحيح: الطابور يفضل أن يتوقف على أن يقبل رسائل قد تضيع. وبعد أن أعدنا تشغيل rmq2 عادت الأغلبية وعاد الطابور يعمل بقائد جديد، ثم أعدنا rmq3:

docker compose up -d --wait rmq2
docker exec lab-rabbitmq-rmq1 rabbitmqctl -q list_queues name leader messages
docker compose up -d --wait rmq3

jobs	rabbit@rmq2	0

ولاحظ أن عدد الرسائل صفر، أي أن الرسائل الخمسمئة التي لم يصل تأكيدها لم تحفظ، والناشر يعرف ذلك لأنه لم يستلم أي تأكيد، فيستطيع أن يعيد إرسالها عندما يعود العنقود.

أثر ال Prefetch على عدالة التوزيع

شرحنا ال Prefetch في مقال المفاهيم بتجربة على Workers متساويين في السرعة، أما هنا فنرى ما يحدث عندما يكون أحدهما أبطأ بعشر مرات، وهي الحالة الواقعية عندما يعمل Worker على سيرفر مزدحم. السكربت fairness.py ينشر 40 مهمة، ويشغل Worker بطيئاً يأخذ نصف ثانية لكل مهمة ويتصل أولاً، وآخر سريعاً يأخذ 50 ملي ثانية ويتصل بعده بنصف ثانية:

# Two consumers on one queue, one of them 10x slower: who gets how many messages, and when does the work end?
import sys, threading, time
from common import connect

prefetch, total = int(sys.argv[1]), 40          # prefetch 0 = no limit
conn, _ = connect()
ch = conn.channel()
ch.queue_declare("tasks", durable=True, arguments={"x-queue-type": "quorum"})
ch.queue_purge("tasks")
ch.confirm_delivery()
for i in range(total):
    ch.basic_publish("", "tasks", str(i).encode())
conn.close()

done, last, start = {"fast": 0, "slow": 0}, {}, time.time()

def worker(name, seconds):
    c, _ = connect()
    w = c.channel()
    w.basic_qos(prefetch_count=prefetch)
    def handle(chan, m, p, body):
        time.sleep(seconds)                       # the "work"
        chan.basic_ack(m.delivery_tag)
        done[name] += 1
        last[name] = time.time() - start
    w.basic_consume("tasks", handle)
    while sum(done.values()) < total:
        c.process_data_events(0.1)
    c.close()

ts = [threading.Thread(target=worker, args=("slow", 0.5)), threading.Thread(target=worker, args=("fast", 0.05))]
ts[0].start(); time.sleep(0.5); ts[1].start()      # the slow worker happens to connect first
for t in ts: t.join()
print(f"prefetch={prefetch or 'unlimited'}: fast got {done['fast']}, slow got {done['slow']}; "
      f"fast idle after {last.get('fast', 0):.1f}s, all done after {max(last.values()):.1f}s")

for p in 0 1 10; do docker compose run --rm -T app python fairness.py $p; done

والمخرج سوف يكون كما يلي:

prefetch=unlimited: fast got 0, slow got 40; fast idle after 0.0s, all done after 20.1s
prefetch=1: fast got 34, slow got 6; fast idle after 3.0s, all done after 3.2s
prefetch=10: fast got 28, slow got 12; fast idle after 2.0s, all done after 6.1s

في المخرج أعلاه لاحظ التالي:

• دون حد لل Prefetch أخذ ال Worker البطيء المهام الأربعين كلها إلى ذاكرته لحظة اتصاله، فلم يجد السريع أي مهمة، وانتهى العمل بعد 20 ثانية، وهي مدة العمل كله على ال Worker البطيء وحده.
• مع القيمة 1 أخذ السريع 34 مهمة لأنه يطلب مهمة جديدة كلما انتهى، وانتهى كل شيء في 3.2 ثوان، أي أسرع بست مرات.
• مع القيمة 10 أخذ البطيء عشر مهام دفعة واحدة ثم انشغل بها، فانتهى السريع بعد ثانيتين وبقي ينتظر بينما البطيء يكمل مهامه حتى الثانية السادسة. لذلك اجعل ال Prefetch صغيراً عندما تختلف سرعة ال Workers أو تطول المهمة، واجعله كبيراً فقط للرسائل الصغيرة السريعة.

واجهة الإدارة Management UI

واجهة الإدارة إضافة مفعلة في ال Images التي تنتهي ب management، وتفتحها في المتصفح على http://127.0.0.1:18672 بالمستخدم lab، والصورة التالية تبين صفحة البداية بعد التجربة، وفي أسفلها العقد الثلاث مع ذاكرة كل عقدة وحد الذاكرة الذي يوقف عنده الوسيط الناشرين:

وصفحة Queues and Streams تبين كل طابور ونوعه والعقدة التي تقوده، والرمز +2 بجانب اسم العقدة يعني أن للطابور نسختين أخريين على عقدتين أخريين:

ولاحظ أن توثيق المراقبة يقول إن الواجهة ليست أداة لحفظ المقاييس على المدى الطويل، فهي للنظر والإدارة اليومية، أما المراقبة فلها قسم خاص بعد قليل. ولا تقم بنشر المنفذ 15672 على الإنترنت، والسبب أن الواجهة تعطي من يدخلها صلاحية حذف الطوابير والرسائل، فاتركها على الشبكة الداخلية أو خلف VPN.

التشغيل في الإنتاج: النسخ الاحتياطي والمراقبة والحماية

• النسخ الاحتياطي Backup: يقول توثيق النسخ الاحتياطي إن الطريقة الموصى بها لحفظ التعريفات Definitions، أي المستخدمين وال Virtual Hosts والموزعات والطوابير والروابط والسياسات، هي تصديرها إلى ملف JSON بالأمر rabbitmqctl export_definitions، أما الرسائل نفسها فلا تؤخذ لها نسخة إلا من عقدة متوقفة، لذلك حمايتها الحقيقية هي النسخ في طوابير Quorum على ثلاث عقد، والأفضل ألا يبقى في الطابور عمل لم يعالج مدة طويلة أصلاً.
• المراقبة Monitoring: فعل الإضافة rabbitmq_prometheus، وهي بحسب توثيق Prometheus تعرض المقاييس على المنفذ 15692 في المسار /metrics، وراقب على الأقل عدد الرسائل الجاهزة في كل طابور، وعدد الرسائل غير المؤكدة، وعدد المستهلكين، وطول ال Dead Letter Queue، وتنبيهات الذاكرة والقرص.
• حد الذاكرة والقرص: بحسب توثيق الذاكرة تستخدم العقدة افتراضياً قرابة 60% من ذاكرة الجهاز، وعندما تتجاوزها توقف كل الاتصالات التي تنشر حتى تنخفض الذاكرة، وهذا ما رأيته في الواجهة حداً قدره 19 جيجابايت على جهاز فيه 31 جيجابايت، لذلك ضع العقدة على سيرفر لا يشاركها فيه تطبيق يأكل الذاكرة، وإلا توقف النشر في أسوأ وقت.
• الحماية Security: مستخدم لكل تطبيق بصلاحيات على ال Virtual Host الخاص به فقط، ولا تستخدم المستخدم guest، وفعل TLS على اتصالات العملاء إذا كانت تمر على شبكة لا تثق بها، كما شرحنا الشهادات في مقال الشهادات الرقمية.
• الترقية Upgrade: كما ذكرنا في قسم الرخصة، دعم المجتمع للسلسلة القديمة ينتهي بسرعة، لذلك اختبر الترقية في بيئة تجربة وخطط لها كل بضعة أشهر، وعنقود من ثلاث عقد يسمح بالترقية عقدة بعد أخرى دون توقف الطوابير.

أخطاء شائعة

• النشر دون تأكيد: رأيت في التجربة 500 رسالة ضاعت دون أن يعرف الناشر، فاستخدم تأكيد النشر في كل رسالة مهمة، وانتظر التأكيدات على دفعات حتى لا تخسر السرعة.
• طوابير Classic في عنقود: الطابور الكلاسيكي يعيش على عقدة واحدة، فالعنقود لا يحميه، واستخدم Quorum لكل طابور يحمل عملاً مهماً.
• عدد زوجي من العقد: عقدتان لا تتحملان سقوط أي عقدة في طابور Quorum، وأربع عقد تتحمل عقدة واحدة فقط مثل ثلاث، فاختر ثلاثاً أو خمساً.
• الاعتماد على أن الرسائل غير الموجهة تظهر كخطأ: الرسالة التي لا تجد رابطاً تحذف بصمت، فاستخدم mandatory أو Alternate Exchange.
• أنماط Topic بنجمة واحدة لمفتاح يكبر: كما رأينا، order.*.sa لا يطابق مفتاحاً من أربع كلمات، فصمم المفاتيح من البداية وثق شكلها.
• Prefetch بلا حد مع Workers مختلفي السرعة: Worker بطيء واحد قد يأخذ كل الرسائل ويترك البقية بلا عمل، كما رأينا في التجربة.
• البقاء على سلسلة قديمة: بعد انتهاء دعم المجتمع لا تصل إصلاحات الأمان مجاناً، فرق إلى آخر سلسلة بانتظام.

حذف المختبر

الأوامر التالية تحذف ال Containers وال Volumes والشبكة التي أنشأتها التجربة، ثم ال Image التي بنيناها:

docker compose --profile tools down -v
docker image rm lab-rabbitmq-app:1

الخلاصة

وصلنا لنهاية الموضوع، وأهم ما فيه:

• RabbitMQ وسيط رسائل برخصة MPL 2.0 يناسب المهام الخلفية والتوجيه المرن، وهو اليوم تحت Broadcom، ودعم المجتمع فيه لآخر سلسلة فقط، فخطط للترقية بانتظام.
• المنتج ينشر إلى موزع، والروابط تقرر الطوابير: Direct بالمفتاح نفسه، وTopic بالنمط حيث * كلمة واحدة و# أي عدد، وFanout للجميع، والرسالة التي لا تجد طابوراً تحذف بصمت ما لم تستخدم mandatory أو Alternate Exchange.
• طوابير Quorum على ثلاث عقد تتحمل سقوط عقدة واحدة، والتأكيد فيها يعني أن الأغلبية حفظت الرسالة، وفي تجربتنا لم تضع أي رسالة من عشرين ألفاً عند إيقاف العقدة القائدة مع التأكيد، وضاعت 500 دونه.
• إعادة الإرسال بعد انقطاع الاتصال تنتج تكراراً، 272 رسالة في التجربة، لذلك يجب أن يكون المستهلك آمناً عند التكرار.
• فقدان الأغلبية يوقف الطابور بدل أن يقبل رسائل قد تضيع، وال Prefetch الصغير يمنع Worker بطيئاً من أخذ كل العمل.
• احفظ التعريفات بالتصدير إلى JSON، وراقب الطوابير عبر Prometheus، ولا تنشر واجهة الإدارة على الإنترنت.

وفي المقالات التالية من المرحلة نشغل Apache Kafka بثلاث عقد ونرى كيف يختلف النسخ فيه، ثم Redpanda وNATS.

دليلالطابور Queue للمبتدئين: لماذا تحتاجه وكيف يعمللماذا تضع العمل البطيء في طابور بدل أن ينتظره المستخدم، وكيف تضمن ألا تضيع الرسالة إذا توقف ال Worker فجأة.دليلالطابور Queue أم بث الأحداث Event Streaming؟ما يحدث للرسالة بعد قراءتها هو الفرق بين طابور مثل RabbitMQ وسجل أحداث مثل Kafka، ومنه تعرف أيهما يناسب مشروعك.دليلIdempotency: كيف تجعل العملية آمنة عند تكرارهالماذا تصل الرسالة نفسها أكثر من مرة، وكيف تكتب مستهلكاً يطبقها مرة واحدة فقط مهما تكررت.دليلالأنظمة الموزعة للمبتدئين: ماذا يحدث عندما تنقطع الشبكةلماذا تتصرف الأنظمة الموزعة بشكل غريب عند انقطاع الشبكة، وكيف تتفق السيرفرات على قرار واحد رغم ذلك.

سجل التحديثات

• أكتوبر 2026: كتابة المقال، وتنفيذ تجربة عنقود RabbitMQ 4.3.6 من ثلاث عقد مع Erlang 27.3.4.18، ومكتبتي pika 1.4.4 وaio-pika 10.1.1 على Python 3.13.16 في Docker.