دليل برمجي لبناء نظام معالجة مهام خلفية باستخدام Huey وSQLite

بقلم: فريق تحرير العلامة التقنية التصنيف: تطوير الويب ١٨ أبريل ٢٠٢٦

في هذا الدليل، سنستعرض كيفية بناء نظام معالجة مهام خلفية فعّال باستخدام مكتبة Huey مباشرة، دون الاعتماد على Redis. ستركز هذه المقالة على إعداد نسخة مدعومة من SQLite لـ Huey، بدء مستهلك حقيقي في بيئة النوتبوك، وتنفيذ أنماط متقدمة من المهام، بما في ذلك محاولات إعادة التشغيل، الأولويات، جدولة المهام، خطوط الأنابيب، التحكم في التزامن، والمراقبة عبر الإشارات.

إعداد بيئة العمل

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

!pip -q install -U huey
import os
import time
from huey import SqliteHuey
DB_PATH = "/content/huey_demo.db"
if os.path.exists(DB_PATH): os.remove(DB_PATH)
huey = SqliteHuey(name="colab-huey", filename=DB_PATH, results=True, store_none=False, utc=True)
print("Huey backend:", type(huey).__name__)
print("SQLite DB at:", DB_PATH)

تسجيل الأحداث ومراقبة المهام

نقوم بتنفيذ معالج للإشارات لتسجيل جميع أحداث دورة حياة المهام في سجل منظم. نتتبع تفاصيل التنفيذ، بما في ذلك معرفات المهام، المعلمات، والاستثناءات، لتعزيز القدرة على المراقبة. من خلال هذه الآلية، نبني نظام مراقبة في الوقت الحقيقي داخل نظامنا غير المتزامن.

EVENT_LOG = []
@huey.signal()
def _log_all_signals(signal, task, exc=None):
    EVENT_LOG.append({
        "ts": datetime.utcnow().isoformat() + "Z",
        "signal": str(signal),
        "task": getattr(task, "name", None),
        "id": getattr(task, "id", None),
        "args": getattr(task, "args", None),
        "kwargs": getattr(task, "kwargs", None),
        "exc": repr(exc) if exc else None,
    })

تعريف المهام المختلفة

نقوم بتعريف عدة مهام مع تخصيص الأولويات، تكوينات إعادة المحاولة، والوعي بالسياق. سنقوم بمحاكاة أحمال عمل مختلفة، بما في ذلك العمليات الحسابية البسيطة، تأخيرات الإدخال/الإخراج، الفشل العرضي، وحسابات وحدة المعالجة المركزية. من خلال ذلك، نوضح كيف يتعامل Huey مع الموثوقية، ترتيب التنفيذ، وبيانات المهام.

@huey.task(priority=50)
def quick_add(a, b):
    return a + b
@huey.task(priority=10)
def slow_io(seconds=1.0):
    time.sleep(seconds)
    return f"slept={seconds}"
@huey.task(retries=3, retry_delay=1, priority=100)
def flaky_network_call(p_fail=0.6):
    if random.random() < p_fail:
        raise RuntimeError("Transient failure (simulated)")
    return "OK"

تنفيذ الإغلاق ومنع التزامن

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

@huey.lock_task("demo:daily-sync")
@huey.task()
def locked_sync_job(tag="sync"):
    time.sleep(1.0)
    return f"locked-job-done:{tag}:{datetime.utcnow().isoformat()}Z"

إعداد مستهلك المهام

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

consumer = huey.create_consumer(workers=4, worker_type=WORKER_THREAD, periodic=True)
consumer_thread = threading.Thread(target=consumer.run, daemon=True)
consumer_thread.start()

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