كيفية حذف الرسائل بشكل انتقائي من قائمة انتظار AMQP (RabbitMQ)؟

Question

أرغب في حذف الرسائل بشكل انتقائي من قائمة انتظار AMQP دون قراءتها.

السيناريو كما يلي:

يرغب Sending Side في انتهاء رسائل النوع X استنادًا إلى حقيقة أن المعلومات الجديدة من النوع X قد وصلت. نظرًا لأنه من المحتمل جدًا ألا يستهلك المشترك آخر رسالة من النوع X حتى الآن ، يجب على الناشر فقط حذف رسائل X-Type السابقة ووضع أحدث رسالة في قائمة الانتظار. يجب أن تكون العملية بأكملها شفافة للمشترك - في الواقع يجب عليه استخدام شيء بسيط مثل Stomp للحصول على الرسائل.

كيف تفعل ذلك باستخدام AMQP؟ أو ربما يكون أكثر ملاءمة في بروتوكول مراسلة آخر؟

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

سيستخدم عميل الناشر روبي ، لكنني في الواقع سأتعامل مع أي لغة بمجرد اكتشاف كيفية القيام بذلك في البروتوكول.

Answer 1

لا يمكنك القيام بذلك حاليًا في RabbitMQ (أو بشكل عام ، في AMQP) تلقائيًا. ولكن ، إليك حل سهل.

لنفترض أنك تريد إرسال ثلاثة أنواع من الرسائل: XS و YS و ZS. إذا فهمت سؤالك بشكل صحيح ، عند وصول رسالة X ، فأنت تريد أن ينسى الوسيط جميع الرسائل X الأخرى التي لم يتم تسليمها.

هذا سهل إلى حد ما في RabbitMQ:

• يعلن المنتج ثلاث قوائم: x و y و z (يتم ربطها تلقائيًا بالتبادل الافتراضي مع أسمائهم كمفاتيح توجيه ، وهو بالضبط ما نريده) ،
• عند نشر رسالة ، يقوم المنتج أولاً بتطهير قائمة الانتظار ذات الصلة (لذلك ، إذا كان ينشر رسالة X ، فإنه يقوم أولاً بتطهير قائمة انتظار X) ؛ هذا يزيل بشكل فعال الرسائل القديمة ،
• يستهلك المستهلك ببساطة من قائمة الانتظار التي يريدها (x للرسائل x ، y للرسائل y ، إلخ) ؛ من وجهة نظره ، يتعين عليه فقط القيام بعمل أساسي للحصول على الرسالة التالية ذات الصلة.

هذا يعني وجود حالة سباق عندما يرسل منتجان نفس النوع من الرسائل في نفس الوقت تقريبًا. والنتيجة هي أنه من الممكن أن يكون لدى قائمة الانتظار رسائل (أو أكثر) في نفس الوقت ، ولكن نظرًا لأن عدد الرسائل يرتدونها في أعلى عدد المنتجين ، وبما أن الرسائل الزائدة مسطحة في النشر التالي ، لا ينبغي أن يكون هذا مشكلة كبيرة.

لتلخيص ، يحتوي هذا الحل على خطوة إضافية واحدة فقط من الحل الأمثل ، وهي Purge Queue X قبل نشر رسالة من النوع X.

إذا كنت بحاجة إلى أي مساعدة في إعداد هذا التكوين ، فإن المكان المثالي لطلب النصيحة هو القائمة البريدية لـ RabbitMQ-Discuss.

Answer 2

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

كل نوع رسالة هو مفتاح. عندما تكتب رسالة جديدة مع نفس المفتاح ، فإنه يكتب القيمة السابقة حتى لا يتمكن قارئ قاعدة البيانات هذه من الحصول على معلومات عن التاريخ.

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

ربما سأفعل شيئًا كهذاkey: typecode value: lastUpdated, important data

ثم سأرسل رسائل تحتويtypecode, lastUpdated وبهذه الطريقة ، يمكن للقارئ مقارنته عن هذا المفتاح إلى هذا المفتاح الذي قرأوه آخر مرة من قاعدة البيانات وتخطي قراءته لأنهم محدثون بالفعل.

إذا كنت بحاجة حقًا إلى القيام بذلك باستخدام AMQP ، فاستخدم RabbitMQ ونوع التبادل المخصص ، وتحديداً تبادل ذاكرة التخزين المؤقت في القيمة الأخيرة. رمز المثال هنا https://github.com/squaremo/rabbitmq-lvc-plugin

Answer 3

يبدو أنه يعمل أيضًا من RabbitMQ Web-Ui ، إذا كنت ترغب فقط في إزالة رسائل N N من قائمة الانتظار

• حدد قائمة الانتظار من علامة التبويب "قوائم الانتظار" ، ثم قم بالتمرير لأسفل إلى القسم "احصل على الرسائل"
• اضبط المعلمة "requeue = لا" وعدد الرسائل التي تريد إزالتها من قائمة الانتظار
• اضغط على زر "Get Messages"

Answer 4

هذا السؤال له رؤية عالية بسبب عنوان ذلك. يتجاوز الوصف يسكن سيناريو أكثر تحديداً. لذلك بالنسبة لأولئك المستخدمين الذين يتطلعون إلى حذف رسالة (تذكر FIFO) التالية من قائمة الانتظار ، يمكنك الاستفادة من RabbitmqAdmin وإصدار الأمر أدناه:

rabbitmqadmin get queue=queuename requeue=false count=1

يستهلك هذا الأمر الرسالة بشكل أساسي ولا يفعل شيئًا. قد يبدو الأمر الكامل مع Flag لأخذ النسخة الاحتياطية للرسالة (الرسائل) كأحد الأطراف أدناه. تأكد من إضافة أي معلمات أخرى وفقًا لمتطلباتك.

sudo python rabbitmqadmin -V virtualhostname -u user -p pass get queue=queuename requeue=false count=1 payload_file=~/origmsg