محرك جدول RabbitMQ

يتيح هذا المحرك تكامل ClickHouse مع RabbitMQ.

يتيح لك RabbitMQ ما يلي:

• نشر تدفقات البيانات أو الاشتراك بها.
• معالجة التدفقات بمجرد توفرها.

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1],
    name2 [type2],
    ...
) ENGINE = RabbitMQ SETTINGS
    rabbitmq_host_port = 'host:port' [or rabbitmq_address = 'amqp(s)://guest:guest@localhost/vhost'],
    rabbitmq_exchange_name = 'exchange_name',
    rabbitmq_format = 'data_format'[,]
    [rabbitmq_exchange_type = 'exchange_type',]
    [rabbitmq_routing_key_list = 'key1,key2,...',]
    [rabbitmq_secure = 0,]
    [rabbitmq_schema = '',]
    [rabbitmq_num_consumers = N,]
    [rabbitmq_num_queues = N,]
    [rabbitmq_queue_base = 'queue',]
    [rabbitmq_persistent = 0,]
    [rabbitmq_skip_broken_messages = N,]
    [rabbitmq_max_block_size = N,]
    [rabbitmq_flush_interval_ms = N,]
    [rabbitmq_queue_settings_list = 'x-dead-letter-exchange=my-dlx,x-max-length=10,x-overflow=reject-publish',]
    [rabbitmq_queue_consume = false,]
    [rabbitmq_address = '',]
    [rabbitmq_vhost = '/',]
    [rabbitmq_username = '',]
    [rabbitmq_password = '',]
    [rabbitmq_commit_on_select = false,]
    [rabbitmq_max_rows_per_message = 1,]
    [rabbitmq_handle_error_mode = 'default']

المعلمات المطلوبة:

• rabbitmq_host_port – المضيف:المنفذ (على سبيل المثال، localhost:5672).
• rabbitmq_exchange_name – اسم الـ exchange في RabbitMQ.
• rabbitmq_format – تنسيق الرسالة. يستخدم الاصطلاح نفسه المستخدم في دالة SQL ‏FORMAT، مثل JSONEachRow. لمزيد من المعلومات، راجع قسم التنسيقات.

المعلمات الاختيارية:

• rabbitmq_exchange_type – نوع RabbitMQ exchange: ‏direct وfanout وtopic وheaders وconsistent_hash. القيمة الافتراضية: fanout.
• rabbitmq_routing_key_list – قائمة بمفاتيح التوجيه مفصولة بفواصل.
• rabbitmq_schema – معلمة يجب استخدامها إذا كانت format تتطلب تعريف schema. على سبيل المثال، يتطلب Cap'n Proto مسار ملف schema واسم الكائن الجذر schema.capnp:Message.
• rabbitmq_num_consumers – عدد الـ المستهلكين لكل جدول. حدّد عددًا أكبر من الـ المستهلكين إذا كان throughput الخاص بـ مستهلك واحد غير كافٍ. القيمة الافتراضية: 1
• rabbitmq_num_queues – العدد الإجمالي لـ قوائم الانتظار. قد تؤدي زيادة هذا العدد إلى تحسين الأداء بشكل ملحوظ. القيمة الافتراضية: 1.
• rabbitmq_queue_base - حدّد بادئة لأسماء الـ قوائم الانتظار. ترد حالات استخدام هذا الإعداد أدناه.
• rabbitmq_persistent - إذا ضُبطت على 1 (true)، فسيُضبط وضع تسليم insert query على 2 (لوضع علامة على الرسائل بأنها 'persistent'). القيمة الافتراضية: 0.
• rabbitmq_skip_broken_messages – مستوى تحمّل محلل رسائل RabbitMQ للرسائل غير المتوافقة مع schema لكل كتلة. إذا كانت rabbitmq_skip_broken_messages = N فإن الـ engine يتخطى N من رسائل RabbitMQ التي يتعذر parse لها (الرسالة الواحدة تعادل row بيانات واحدة). القيمة الافتراضية: 0.
• rabbitmq_max_block_size - عدد الـ row التي تُجمع قبل flush البيانات من RabbitMQ. القيمة الافتراضية: max_insert_block_size.
• rabbitmq_flush_interval_ms - timeout الخاص بعملية flush البيانات من RabbitMQ. القيمة الافتراضية: stream_flush_interval_ms.
• rabbitmq_queue_settings_list - يتيح ضبط إعدادات RabbitMQ عند إنشاء قائمة انتظار. الإعدادات المتاحة: x-max-length وx-max-length-bytes وx-message-ttl وx-expires وx-priority وx-max-priority وx-overflow وx-dead-letter-exchange وx-queue-type. يتم تمكين الإعداد durable تلقائيًا للـ قائمة انتظار.
• rabbitmq_address - عنوان الـ اتصال: amqp(s)://user:password@host:port/vhost. استخدم هذا الإعداد أو rabbitmq_host_port؛ وإذا تم ضبط كليهما، فسيُستخدم rabbitmq_address. يتم التحقق من مضيفه ومنفذه مقابل remote_url_allow_hosts.
• rabbitmq_vhost - ‏RabbitMQ vhost. القيمة الافتراضية: '/'.
• rabbitmq_queue_consume - استخدم قوائم انتظار معرّفة من قبل المستخدم، ولا تُجرِ أي إعداد لـ RabbitMQ: تعريف exchanges وقوائم الانتظار وbindings. القيمة الافتراضية: false.
• rabbitmq_username - اسم مستخدم RabbitMQ.
• rabbitmq_password - كلمة مرور RabbitMQ.
• reject_unhandled_messages - ارفض الرسائل (أرسل إقرارًا سلبيًا إلى RabbitMQ) عند حدوث error. يتم تمكين هذا الإعداد تلقائيًا إذا كان هناك x-dead-letter-exchange معرّف في rabbitmq_queue_settings_list.
• rabbitmq_commit_on_select - نفّذ commit للرسائل عند تنفيذ select query. القيمة الافتراضية: false.
• rabbitmq_max_rows_per_message — الحد الأقصى لعدد الـ صفوف المكتوبة في رسالة RabbitMQ واحدة للتنسيقات المعتمدة على الصفوف. القيمة الافتراضية: 1.
• rabbitmq_empty_queue_backoff_start_ms — نقطة بداية backoff لإعادة جدولة القراءة إذا كانت RabbitMQ قائمة انتظار فارغة.
• rabbitmq_empty_queue_backoff_end_ms — نقطة نهاية backoff لإعادة جدولة القراءة إذا كانت RabbitMQ قائمة انتظار فارغة.
• rabbitmq_empty_queue_backoff_step_ms — قيمة خطوة backoff لإعادة جدولة القراءة إذا كانت RabbitMQ قائمة انتظار فارغة.
• rabbitmq_handle_error_mode — كيفية التعامل مع error في محرك RabbitMQ. القيم الممكنة: default (سيتم طرح الاستثناء إذا فشلنا في parse رسالة)، stream (سيتم حفظ رسالة الاستثناء والرسالة الخام في أعمدة افتراضية _error و_raw_message)، dead_letter_queue (سيتم حفظ البيانات المرتبطة بالخطأ في system.dead_letter_queue).

باستخدام صيغة rabbitmq_host_port، اضبط rabbitmq_secure = 1 لاستخدام TLS. أما باستخدام صيغة rabbitmq_address، فيُحدَّد النقل من مخطط عنوان URI، لذا استخدم amqps: rabbitmq_address = 'amqps://guest:guest@localhost/vhost'. يتم تجاهل rabbitmq_secure لصيغة العنوان، ويُرفض استخدام rabbitmq_secure = 1 مع عنوان amqp:// بنص عادي، بدلًا من الاتصال بصمت دون تشفير. السلوك الافتراضي للمكتبة المستخدمة هو عدم التحقق مما إذا كان اتصال TLS المُنشأ آمنًا بما يكفي. فسواء كانت الشهادة منتهية الصلاحية أو موقَّعة ذاتيًا أو مفقودة أو غير صالحة، يُسمح بالاتصال بكل بساطة. وقد يُطبَّق في المستقبل تحقّق أكثر صرامة من الشهادات.

يمكن أيضًا إضافة إعدادات التنسيق إلى جانب الإعدادات المتعلقة بـ rabbitmq.

مثال:

CREATE TABLE queue (
    key UInt64,
    value UInt64,
    date DateTime
  ) ENGINE = RabbitMQ SETTINGS rabbitmq_host_port = 'localhost:5672',
                            rabbitmq_exchange_name = 'exchange1',
                            rabbitmq_format = 'JSONEachRow',
                            rabbitmq_num_consumers = 5,
                            date_time_input_format = 'best_effort';

يجب إضافة تهيئة خادم RabbitMQ باستخدام ملف إعدادات ClickHouse.

التهيئة المطلوبة:

<rabbitmq>
    <username>root</username>
    <password>clickhouse</password>
</rabbitmq>

تهيئة إضافية:

<rabbitmq>
    <vhost>clickhouse</vhost>
</rabbitmq>

لا يُعد SELECT مفيدًا كثيرًا لقراءة الرسائل (إلا لأغراض Debug)، لأن كل رسالة لا يمكن قراءتها إلا مرة واحدة. والأكثر عملية هو إنشاء مسارات في الوقت الفعلي باستخدام العروض المادية. وللقيام بذلك:

1. استخدم المحرّك لإنشاء مستهلك RabbitMQ واعتبره تدفق بيانات.
2. أنشئ جدولًا بالبنية المطلوبة.
3. أنشئ عرضًا ماديًا يحوّل البيانات من المحرّك ويضعها في جدول أُنشئ مسبقًا.

عند ربط MATERIALIZED VIEW بالمحرّك، يبدأ في جمع البيانات في الخلفية. يتيح لك ذلك الاستمرار في تلقي الرسائل من RabbitMQ وتحويلها إلى التنسيق المطلوب باستخدام SELECT. يمكن لجدول RabbitMQ واحد أن يحتوي على أي عدد تريده من العروض المادية.

يمكن توجيه البيانات استنادًا إلى rabbitmq_exchange_type وrabbitmq_routing_key_list المحددَين. لا يمكن أن يوجد أكثر من exchange واحد لكل جدول. ويمكن مشاركة exchange واحد بين عدة جداول، مما يتيح التوجيه إلى عدة جداول في الوقت نفسه.

خيارات نوع exchange:

• direct - يعتمد التوجيه على المطابقة التامة للمفاتيح. مثال على قائمة مفاتيح الجدول: key1,key2,key3,key4,key5، ويمكن أن يطابق مفتاح الرسالة أيًا منها.
• fanout - توجيه إلى جميع الجداول (حيث يكون اسم exchange نفسه) بغض النظر عن المفاتيح.
• topic - يعتمد التوجيه على patterns بمفاتيح مفصولة بنقاط. أمثلة: *.logs, records.*.*.2020, *.2018,*.2019,*.2020.
• headers - يعتمد التوجيه على مطابقات key=value مع الإعداد x-match=all أو x-match=any. مثال على قائمة مفاتيح الجدول: x-match=all,format=logs,type=report,year=2020.
• consistent_hash - تُوزَّع البيانات بالتساوي بين جميع الجداول المرتبطة (حيث يكون اسم exchange نفسه). لاحظ أن نوع exchange هذا يجب تمكينه عبر إضافة RabbitMQ: rabbitmq-plugins enable rabbitmq_consistent_hash_exchange.

يمكن استخدام الإعداد rabbitmq_queue_base في الحالات التالية:

• للسماح لجداول مختلفة بمشاركة قوائم الانتظار، بحيث يمكن تسجيل عدة مستهلكين لقوائم الانتظار نفسها، مما يحسن الأداء. وعند استخدام الإعدادين rabbitmq_num_consumers و/أو rabbitmq_num_queues، تتحقق المطابقة التامة لقوائم الانتظار إذا كانت هذه المعلمات متماثلة.
• لإتاحة استعادة القراءة من قوائم انتظار durable محددة عندما لا تكون كل الرسائل قد استُهلكت بنجاح. لاستئناف الاستهلاك من قائمة انتظار واحدة محددة، اضبط اسمها في الإعداد rabbitmq_queue_base ولا تحدد rabbitmq_num_consumers وrabbitmq_num_queues (القيمة الافتراضية هي 1). ولاستئناف الاستهلاك من جميع قوائم الانتظار التي أُعلنت لجدول محدد، ما عليك سوى تحديد الإعدادات نفسها: rabbitmq_queue_base وrabbitmq_num_consumers وrabbitmq_num_queues. افتراضيًا، ستكون أسماء قوائم الانتظار فريدة لكل جدول.
• لإعادة استخدام قوائم الانتظار لأنها مُعلنة كـ durable ولا تُحذف تلقائيًا. (يمكن حذفها عبر أي أداة من أدوات RabbitMQ CLI.)

لتحسين الأداء، تُجمَّع الرسائل المستلمة في blocks بحجم max_insert_block_size. وإذا لم تتشكّل الكتلة خلال stream_flush_interval_ms مللي ثانية، فستُفرَّغ البيانات إلى الجدول بغض النظر عن اكتمال الكتلة.

إذا تم تحديد الإعدادين rabbitmq_num_consumers و/أو rabbitmq_num_queues مع rabbitmq_exchange_type، فعندئذٍ:

• يجب تمكين إضافة rabbitmq-consistent-hash-exchange.
• يجب تحديد الخاصية message_id للرسائل المنشورة (فريدة لكل رسالة/Batch).

بالنسبة إلى insert query، توجد بيانات وصفية للرسالة تُضاف لكل رسالة منشورة: messageID والعَلم republished (تكون قيمته true إذا نُشرت أكثر من مرة) — ويمكن الوصول إليهما عبر headers الرسالة.

لا تستخدم الجدول نفسه لعمليات insert والعروض المادية.

مثال:

CREATE TABLE queue (
    key UInt64,
    value UInt64
  ) ENGINE = RabbitMQ SETTINGS rabbitmq_host_port = 'localhost:5672',
                            rabbitmq_exchange_name = 'exchange1',
                            rabbitmq_exchange_type = 'headers',
                            rabbitmq_routing_key_list = 'format=logs,type=report,year=2020',
                            rabbitmq_format = 'JSONEachRow',
                            rabbitmq_num_consumers = 5;

CREATE TABLE daily (key UInt64, value UInt64)
    ENGINE = MergeTree() ORDER BY key;

CREATE MATERIALIZED VIEW consumer TO daily
    AS SELECT key, value FROM queue;

SELECT key, value FROM daily ORDER BY key;

• _exchange_name - اسم الـ exchange في RabbitMQ. نوع البيانات: String.
• _channel_id - معرّف القناة ChannelID التي أُعلن عليها المستهلك الذي استلم الرسالة. نوع البيانات: String.
• _delivery_tag - قيمة DeliveryTag للرسالة المستلمة. وهي ضمن نطاق كل قناة. نوع البيانات: UInt64.
• _redelivered - العلامة redelivered الخاصة بالرسالة. نوع البيانات: UInt8.
• _message_id - قيمة messageID للرسالة المستلمة؛ وتكون غير فارغة إذا كان قد جرى تعيينها عند نشر الرسالة. نوع البيانات: String.
• _timestamp - الطابع الزمني للرسالة المستلمة؛ ويكون غير فارغ إذا كان قد جرى تعيينه عند نشر الرسالة. نوع البيانات: UInt64.

أعمدة افتراضية إضافية عندما يكون rabbitmq_handle_error_mode='stream':

• _raw_message - الرسالة الخام التي تعذّر تحليلها بنجاح. نوع البيانات: Nullable(String).
• _error - رسالة الاستثناء التي حدثت أثناء فشل التحليل. نوع البيانات: Nullable(String).

ملاحظة: لا تُملأ الأعمدة الافتراضية _raw_message و _error إلا عند حدوث استثناء أثناء التحليل، وتكون دائمًا NULL عندما تُحلَّل الرسالة بنجاح.

على الرغم من أنه يمكنك تحديد تعبيرات الأعمدة الافتراضية (مثل DEFAULT وMATERIALIZED وALIAS) في تعريف الجدول، فسيتم تجاهلها. وبدلاً من ذلك، ستُملأ الأعمدة بالقيم الافتراضية المقابلة لأنواعها.

يدعم محرك RabbitMQ جميع التنسيقات التي يدعمها ClickHouse. يعتمد عدد الصفوف في رسالة RabbitMQ الواحدة على ما إذا كان التنسيق يعتمد على الصفوف أو على الكتل:

• بالنسبة إلى التنسيقات المعتمدة على الصفوف، يمكن التحكم في عدد الصفوف في رسالة RabbitMQ الواحدة من خلال ضبط rabbitmq_max_rows_per_message.
• بالنسبة إلى التنسيقات المعتمدة على الكتل، لا يمكن تقسيم الكتلة إلى أجزاء أصغر، ولكن يمكن التحكم في عدد الصفوف في الكتلة الواحدة باستخدام الإعداد العام max_block_size.

قد يفقد محرك RabbitMQ بصمت صفوفًا سبق استهلاكها إذا فُقدت ذاكرة التخزين المؤقت للصفحات في نظام التشغيل قبل كتابة البيانات المُدرجة على القرص. بعد دفع دفعة إلى العروض المادية التابعة، يرسل المستهلك basic.ack إلى الوسيط، ما يتيح للوسيط حذف تلك الرسائل. لكن الصفوف المُدرجة لا تصبح متينة إلا بعد مزامنة الجزء الهدف مع القرص باستخدام fsync، وهو ما لا يحدث افتراضيًا بصورة متزامنة (fsync_after_insert = 0). إذا فُقدت ذاكرة التخزين المؤقت للصفحات بعد الإقرار وقبل مزامنة الجزء الهدف مع القرص، يكون الوسيط قد حذف الرسائل بالفعل، ويستأنف المستهلك بعد تجاوزها عند إعادة الاتصال، فتُفقد الصفوف دون أي خطأ، ويكون count() أصغر ببساطة. لا يكشف إنهاء العملية العادي عن ذلك، لأن النواة تحتفظ بذاكرة التخزين المؤقت للصفحات وتكتبها إلى القرص في النهاية. أما فقدان ذاكرة التخزين المؤقت للصفحات فيكشف هذه المشكلة؛ ومن أمثلته انقطاع الطاقة على مستوى الجهاز، أو إعادة تشغيل غير نظيفة للمضيف أو للنواة.

بالنسبة إلى مسار الاستهلاك الموصى به عبر العرض المادي (لا يُرسل الإقرار إلا بعد اكتمال مسار الإدراج بالكامل)، فإن تعيين fsync_after_insert = 1 (وfsync_part_directory = 1) على جداول MergeTree الهدف يجعل الأجزاء المُدرجة متينة قبل إرسال الإقرار، مما يضيّق هذه النافذة بدرجة كبيرة. يجب تمكين هذا الإعداد في كل جدول MergeTree تُدرج فيه الدفعة، بما في ذلك أهداف العروض المادية المتسلسلة؛ إذ قد يفقد أي جدول من هذه الجداول يُترك على الإعداد الافتراضي جزأه. لا تكتسب المراحل الوسيطة غير المتزامنة المتانة من هذا الإعداد وحده: فعلى سبيل المثال، يُدرج هدف Distributed في الخلفية عندما يكون distributed_foreground_insert = 0، وهو الإعداد الافتراضي خارج ClickHouse Cloud، لذا يحتاج إلى إعدادات متانة خاصة به أو إلى إدراج متزامن. لا ينطبق هذا التخفيف أيضًا على INSERT ... SELECT ... FROM <rabbitmq_table> مباشر مع rabbitmq_commit_on_select = 1، حيث يُرسَل الإقرار بالرسائل عند بلوغ القراءة نهايتها، بدلًا من إرساله بعد أن تكتب الوجهة جزءًا متينًا.