bull
v4.16.4
قم بتعزيز قوائم الانتظار الخاصة بك بواجهة أمامية احترافية:
احصل على نظرة عامة كاملة على جميع قوائم الانتظار الخاصة بك.
فحص الوظائف أو البحث أو إعادة المحاولة أو الترويج للوظائف المتأخرة.
المقاييس والإحصائيات.
والعديد من الميزات.
قم بالتسجيل في Taskforce.sh
الحد الأدنى من استخدام وحدة المعالجة المركزية (CPU) بسبب التصميم الخالي من الاستقصاء.
تصميم قوي يعتمد على Redis.
وظائف متأخرة.
جدولة المهام وتكرارها وفقًا لمواصفات cron.
محدد معدل للوظائف.
إعادة المحاولة.
أولوية.
التزامن.
إيقاف مؤقت/استئناف — عالميًا أو محليًا.
أنواع وظائف متعددة لكل قائمة انتظار.
وظائف المعالجة المترابطة (وضع الحماية).
الاسترداد التلقائي من تعطل العملية.
والقادم على خارطة الطريق..
إقرار بإكمال المهمة (يمكنك استخدام نمط قائمة انتظار الرسائل في هذه الأثناء).
العلاقات الوظيفية بين الوالدين والطفل.
هناك عدد قليل من واجهات المستخدم الخارجية التي يمكنك استخدامها للمراقبة:
BullMQ
فرقة العمل
الثور v3
فرقة العمل
لوحة الثور
الثور repl
مراقب الثور
مونيتوريو
الثور <= v2
مصارع الثيران
رد فعل الثور
توريرو
مع بروميثيوس الثور قائمة الانتظار المصدر
نظرًا لوجود عدد قليل من حلول قائمة انتظار المهام، إليك جدول يقارن بينها:
| Dragonfly هو بديل جديد لـ Redis™ متوافق تمامًا مع BullMQ ويقدم بعض المزايا المهمة مقارنة بـ Redis™ مثل الأداء الأفضل الهائل من خلال استخدام جميع مراكز وحدة المعالجة المركزية المتاحة وهياكل بيانات أسرع وأكثر كفاءة في الذاكرة. اقرأ المزيد هنا حول كيفية استخدامه مع BullMQ. | |
| إذا كنت بحاجة إلى مثيلات Redis إنتاجية عالية الجودة لمشروع Bull الخاص بك، فيرجى التفكير في الاشتراك في Memetria for Redis، الرائدة في استضافة Redis والتي تعمل بشكل مثالي مع BullMQ. استخدم الرمز الترويجي "BULLMQ" عند التسجيل لمساعدتنا في رعاية تطوير BullMQ! |
| ميزة | BullMQ-Pro | BullMQ | ثور | كوي | نحلة | جدول الأعمال |
|---|---|---|---|---|---|---|
| الخلفية | redis | redis | redis | redis | redis | مونجو |
| يمكن ملاحظتها | ✓ | |||||
| حد سعر المجموعة | ✓ | |||||
| دعم المجموعة | ✓ | |||||
| دعم الدفعات | ✓ | |||||
| تبعيات الوالدين / الطفل | ✓ | ✓ | ||||
| الأولويات | ✓ | ✓ | ✓ | ✓ | ✓ | |
| التزامن | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |
| وظائف متأخرة | ✓ | ✓ | ✓ | ✓ | ✓ | |
| الأحداث العالمية | ✓ | ✓ | ✓ | ✓ | ||
| محدد المعدل | ✓ | ✓ | ✓ | |||
| وقفة / استئناف | ✓ | ✓ | ✓ | ✓ | ||
| عامل رمل | ✓ | ✓ | ✓ | |||
| وظائف متكررة | ✓ | ✓ | ✓ | ✓ | ||
| العمليات الذرية | ✓ | ✓ | ✓ | ✓ | ||
| المثابرة | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |
| واجهة المستخدم | ✓ | ✓ | ✓ | ✓ | ✓ | |
| الأمثل ل | وظائف / رسائل | وظائف / رسائل | وظائف / رسائل | وظائف | رسائل | وظائف |
npm تثبيت الثور --حفظ
أو
الغزل إضافة الثور
المتطلبات: يتطلب Bull إصدار Redis أكبر من أو يساوي 2.8.18 .
تثبيت npm @types/bull --save-dev
إضافة الغزل --dev @types/bull
يتم الاحتفاظ بالتعريفات حاليًا في DefinitelyTyped repo.
نحن نرحب بجميع أنواع المساهمات، سواء إصلاحات التعليمات البرمجية أو الميزات الجديدة أو تحسينات المستندات. يتم فرض تنسيق التعليمات البرمجية بواسطة أجمل. بالنسبة للالتزامات، يرجى اتباع اتفاقية الالتزامات التقليدية. يجب أن تجتاز جميع التعليمات البرمجية قواعد الوبر ومجموعات الاختبار قبل أن يتم دمجها في التطوير.
const Queue = require('bull');const videoQueue = new Queue('تحويل ترميز الفيديو', 'redis://127.0.0.1:6379');const audioQueue = new Queue('تحويل ترميز الصوت', { redis: { منفذ : 6379، المضيف: '127.0.0.1'، كلمة المرور: 'foobared' } }); // حدد اتصال Redis باستخدام objectconst imageQueue = new Queue('image transcoding');const pdfQueue = new Queue('pdf transcoding');videoQueue.process(function (job, Done) {
تحتوي // job.data على البيانات المخصصة التي تم تمريرها عند إنشاء الوظيفة
// job.id يحتوي على معرف هذه الوظيفة.
// تحويل الفيديو بشكل غير متزامن والإبلاغ عن التقدم
job.progress(42);
// تم الاتصال عند الانتهاء
منتهي()؛
// أو إعطاء خطأ إذا كان الخطأ
تم (خطأ جديد ("خطأ في تحويل الشفرة"))؛
// أو تمريرها نتيجة
تم (خالي، { معدل الإطارات: 29.5 /* إلخ... */ });
// إذا طرحت المهمة استثناءً لم تتم معالجته، فسيتم التعامل معه أيضًا بشكل صحيح
رمي خطأ جديد ("بعض الأخطاء غير المتوقعة")؛})؛audioQueue.process(function (job, did) {
// تحويل الصوت بشكل غير متزامن والإبلاغ عن التقدم
job.progress(42);
// تم الاتصال عند الانتهاء
منتهي()؛
// أو إعطاء خطأ إذا كان الخطأ
تم (خطأ جديد ("خطأ في تحويل الشفرة"))؛
// أو تمريرها نتيجة
تم (خالي، { معدل العينات: 48000 /* الخ... */ });
// إذا طرحت المهمة استثناءً لم تتم معالجته، فسيتم التعامل معه أيضًا بشكل صحيح
رمي خطأ جديد ("بعض الأخطاء غير المتوقعة")؛})؛ imageQueue.process (وظيفة (مهمة، تم) {
// تحويل الصورة بشكل غير متزامن والإبلاغ عن التقدم
job.progress(42);
// تم الاتصال عند الانتهاء
منتهي()؛
// أو إعطاء خطأ إذا كان الخطأ
تم (خطأ جديد ("خطأ في تحويل الشفرة"))؛
// أو تمريرها نتيجة
تم(خالي, { العرض: 1280, الارتفاع: 720 /* إلخ... */ });
// إذا طرحت المهمة استثناءً لم تتم معالجته، فسيتم التعامل معه أيضًا بشكل صحيح
رمي خطأ جديد ("بعض الأخطاء غير المتوقعة")؛})؛pdfQueue.process(function (job) {
// يمكن للمعالجات أيضًا إرجاع الوعود بدلاً من استخدام رد الاتصال المنجز
return pdfAsyncProcessor();});videoQueue.add({ video: 'http://example.com/video1.mov' });audioQueue.add({ audio: 'http://example.com/audio1.mp3 ' });imageQueue.add({ image: 'http://example.com/image1.tiff' }); وبدلاً من ذلك، يمكنك إرجاع الوعود بدلاً من استخدام رد الاتصال done :
videoQueue.process(function (job) { // لا تنس إزالة رد الاتصال الذي تم إجراؤه!
// ببساطة قم بإرجاع الوعد
return fetchVideo(job.data.url).then(transcodeVideo);
// يعالج رفض الوعد
return Promise.reject(new Error('خطأ في تحويل الشفرة'));
// يمرر القيمة التي تم حل الوعد بها إلى الحدث "المكتمل".
return Promise.resolve({ معدل الإطارات: 29.5 /* إلخ... */ });
// إذا طرحت المهمة استثناءً لم تتم معالجته، فسيتم التعامل معه أيضًا بشكل صحيح
رمي خطأ جديد ("بعض الأخطاء غير المتوقعة")؛
// نفس
return Promise.reject(new Error('بعض الأخطاء غير المتوقعة'));});يمكن أيضًا تشغيل وظيفة العملية في عملية منفصلة. وهذا له العديد من المزايا:
تكون العملية في وضع الحماية، لذا إذا تعطلت فلن يؤثر ذلك على العامل.
يمكنك تشغيل كود الحظر دون التأثير على قائمة الانتظار (لن تتوقف المهام).
استخدام أفضل بكثير لوحدات المعالجة المركزية متعددة النواة.
اتصالات أقل بـ redis.
من أجل استخدام هذه الميزة فقط قم بإنشاء ملف منفصل مع المعالج:
// المعالج.jsmodule.exports = الوظيفة (الوظيفة) {
// قم ببعض الأعمال الشاقة
إرجاع Promise.resolve(result);}وتعريف المعالج كالتالي:
// عملية واحدة:queue.process('/path/to/my/processor.js');// يمكنك استخدام التزامن أيضًا:queue.process(5, '/path/to/my/processor.js' );// والمعالجات المسماة:queue.process('myprocessor', 5, '/path/to/my/processor.js');يمكن إضافة مهمة إلى قائمة الانتظار ومعالجتها بشكل متكرر وفقًا لمواصفات cron:
PaymentsQueue.process(function (job) {// التحقق من المدفوعات
});
// كرر مهمة الدفع مرة واحدة كل يوم الساعة 3:15 (صباحًا)
PaymentsQueue.add(PaymentsData, { Repeat: { cron: '15 3 * * *' } });كنصيحة، تحقق من تعبيراتك هنا للتأكد من صحتها: cron Expression Generator
يمكن إيقاف قائمة الانتظار مؤقتًا واستئنافها عالميًا (تمرير true لإيقاف المعالجة مؤقتًا لهذا العامل فقط):
queue.pause().then(function () {
// تم إيقاف قائمة الانتظار مؤقتًا الآن});queue.resume().then(function () {
// يتم استئناف قائمة الانتظار الآن})تصدر قائمة الانتظار بعض الأحداث المفيدة، على سبيل المثال...
.on('مكتمل'، الوظيفة (الوظيفة، النتيجة) {
// اكتملت المهمة بنتيجة الإخراج!})لمزيد من المعلومات حول الأحداث، بما في ذلك القائمة الكاملة للأحداث التي تم إطلاقها، راجع مرجع الأحداث
قوائم الانتظار رخيصة، لذا إذا كنت بحاجة إلى الكثير منها، فما عليك سوى إنشاء قوائم جديدة بأسماء مختلفة:
const userJohn = قائمة انتظار جديدة('john');const userLisa = قائمة انتظار جديدة('lisa');...ومع ذلك، ستتطلب كل نسخة قائمة انتظار اتصالات redis جديدة، وتحقق من كيفية إعادة استخدام الاتصالات أو يمكنك أيضًا استخدام المعالجات المسماة لتحقيق نتيجة مماثلة.
ملاحظة: من الإصدار 3.2.0 وما فوق، يوصى باستخدام المعالجات المترابطة بدلاً من ذلك.
تتميز قوائم الانتظار بأنها قوية ويمكن تشغيلها بالتوازي في عدة سلاسل أو عمليات دون أي مخاطر أو تلف في قائمة الانتظار. تحقق من هذا المثال البسيط باستخدام المجموعة لموازاة المهام عبر العمليات:
const Queue = require('bull');const block = require('cluster');const numWorkers = 8;const queue = new Queue('اختبار قائمة الانتظار المتزامنة');if (cluster.isMaster) {
for (let i = 0; i < numWorkers; i++) {cluster.fork();
}
cluster.on('online', function (worker) {// لنقم بإنشاء بعض الوظائف لقائمة الانتظارworkersfor (let i = 0; i < 500; i++) { queue.add({ foo: 'bar' }); };
});
cluster.on('exit', function (worker, code, signal) {console.log('worker ' +worker.process.pid + ' die');
})؛} آخر {
queue.process(function (job, jobDone) {console.log('المهمة يقوم بها العامل',cluster.worker.id, job.id);jobDone();
});}للحصول على الوثائق الكاملة، راجع المرجع والأنماط الشائعة:
الدليل - نقطة البداية للتطوير مع Bull.
مرجع - وثيقة مرجعية تحتوي على جميع الكائنات والأساليب المتاحة.
الأنماط - مجموعة من الأمثلة للأنماط الشائعة.
الترخيص - ترخيص الثور - إنه معهد ماساتشوستس للتكنولوجيا.
إذا رأيت أي شيء يمكن أن يستخدم المزيد من المستندات، فيرجى إرسال طلب سحب!
تهدف قائمة الانتظار إلى استراتيجية عمل "مرة واحدة على الأقل". وهذا يعني أنه في بعض الحالات، يمكن معالجة المهمة أكثر من مرة. يحدث هذا غالبًا عندما يفشل العامل في الاحتفاظ بقفل لمهمة معينة خلال المدة الإجمالية للمعالجة.
عندما يقوم عامل بمعالجة مهمة ما، فإنه سيبقي الوظيفة "مقفلة" حتى لا يتمكن العمال الآخرون من معالجتها.
من المهم أن تفهم كيفية عمل القفل لمنع وظائفك من فقدان القفل - وتوقفها - وإعادة تشغيلها نتيجة لذلك. يتم تنفيذ القفل داخليًا عن طريق إنشاء قفل لـ lockDuration على الفاصل الزمني lockRenewTime (والذي يكون عادةً نصف lockDuration ). إذا انقضت lockDuration قبل تجديد القفل، فسيتم اعتبار المهمة متوقفة وسيتم إعادة تشغيلها تلقائيًا؛ سيتم معالجتها بشكل مزدوج . يمكن أن يحدث هذا عندما:
تنتهي عملية العقدة التي تقوم بتشغيل معالج المهمة لديك بشكل غير متوقع.
كان معالج المهمة لديك يستهلك الكثير من وحدة المعالجة المركزية (CPU) وأوقف حلقة حدث Node، ونتيجة لذلك، لم يتمكن Bull من تجديد قفل الوظيفة (راجع رقم 488 للتعرف على كيفية اكتشاف ذلك بشكل أفضل). يمكنك إصلاح ذلك عن طريق تقسيم معالج المهمة الخاص بك إلى أجزاء أصغر بحيث لا يتمكن أي جزء واحد من منع حلقة حدث العقدة. وبدلاً من ذلك، يمكنك تمرير قيمة أكبر لإعداد lockDuration (مع المفاضلة التي ستستغرق وقتًا أطول للتعرف على المهمة المتوقفة الحقيقية).
على هذا النحو، يجب عليك دائمًا الاستماع إلى الحدث stalled وتسجيل ذلك في نظام مراقبة الأخطاء الخاص بك، لأن هذا يعني أنه من المحتمل أن تتم معالجة مهامك بشكل مزدوج.
كإجراء وقائي، لن تتم إعادة تشغيل المهام التي بها مشكلات إلى أجل غير مسمى (على سبيل المثال، إذا كان معالج المهام يتعطل دائمًا عملية العقدة الخاصة به)، فسيتم استرداد المهام من حالة التوقف بحد أقصى maxStalledCount مرات (الافتراضي: 1 ).
