django rq
v3.0
تكامل Django مع RQ، وهي مكتبة انتظار Python مبنية على Redis. Django-RQ هو تطبيق بسيط يسمح لك بتكوين قوائم الانتظار الخاصة بك في settings.py الخاص بـ Django واستخدامها بسهولة في مشروعك.
إذا وجدت أن django-rq مفيد، فيرجى التفكير في دعم تطويره عبر Tidelift.
django-rq (أو تنزيله من PyPI): pip install django - rqdjango_rq إلى INSTALLED_APPS في settings.py : INSTALLED_APPS = (
# other apps
"django_rq" ,
)settings.py الخاص بـ Django: RQ_QUEUES = {
'default' : {
'HOST' : 'localhost' ,
'PORT' : 6379 ,
'DB' : 0 ,
'USERNAME' : 'some-user' ,
'PASSWORD' : 'some-password' ,
'DEFAULT_TIMEOUT' : 360 ,
'REDIS_CLIENT_KWARGS' : { # Eventual additional Redis connection arguments
'ssl_cert_reqs' : None ,
},
},
'with-sentinel' : {
'SENTINELS' : [( 'localhost' , 26736 ), ( 'localhost' , 26737 )],
'MASTER_NAME' : 'redismaster' ,
'DB' : 0 ,
# Redis username/password
'USERNAME' : 'redis-user' ,
'PASSWORD' : 'secret' ,
'SOCKET_TIMEOUT' : 0.3 ,
'CONNECTION_KWARGS' : { # Eventual additional Redis connection arguments
'ssl' : True
},
'SENTINEL_KWARGS' : { # Eventual Sentinel connection arguments
# If Sentinel also has auth, username/password can be passed here
'username' : 'sentinel-user' ,
'password' : 'secret' ,
},
},
'high' : {
'URL' : os . getenv ( 'REDISTOGO_URL' , 'redis://localhost:6379/0' ), # If you're on Heroku
'DEFAULT_TIMEOUT' : 500 ,
},
'low' : {
'HOST' : 'localhost' ,
'PORT' : 6379 ,
'DB' : 0 ,
}
}
RQ_EXCEPTION_HANDLERS = [ 'path.to.my.handler' ] # If you need custom exception handlersdjango_rq.urls في urls.py الخاص بك: urlpatterns += [
path ( 'django-rq/' , include ( 'django_rq.urls' ))
] يتيح لك Django-RQ وضع المهام بسهولة في أي من قوائم الانتظار المحددة في settings.py . لأنه يأتي مع عدد قليل من الوظائف المساعدة:
enqueue - دفع المهمة إلى قائمة الانتظار default : import django_rq
django_rq . enqueue ( func , foo , bar = baz )get_queue - يقوم بإرجاع مثيل قائمة Queue . import django_rq
queue = django_rq . get_queue ( 'high' )
queue . enqueue ( func , foo , bar = baz ) بالإضافة إلى وسيطة name ، تقبل get_queue أيضًا وسيطات default_timeout و is_async و autocommit و connection و queue_class . على سبيل المثال:
queue = django_rq . get_queue ( 'default' , autocommit = True , is_async = True , default_timeout = 360 )
queue . enqueue ( func , foo , bar = baz )يمكنك توفير كائن اتصال Redis المفرد الخاص بك لهذه الوظيفة حتى لا تقوم بإنشاء كائن اتصال جديد لكل تعريف قائمة انتظار. سيساعدك هذا على تحديد عدد الاتصالات بخادم Redis. على سبيل المثال:
import django_rq
import redis
redis_cursor = redis . StrictRedis ( host = '' , port = '' , db = '' , password = '' )
high_queue = django_rq . get_queue ( 'high' , connection = redis_cursor )
low_queue = django_rq . get_queue ( 'low' , connection = redis_cursor )get_connection - يقبل وسيطة اسم قائمة الانتظار الفردية (القيمة الافتراضية هي "افتراضي") ويعيد الاتصال إلى خادم Redis الخاص بقائمة الانتظار: import django_rq
redis_conn = django_rq . get_connection ( 'high' )get_worker - يقبل أسماء قوائم الانتظار الاختيارية ويعيد مثيل RQ Worker الجديد لقوائم الانتظار المحددة (أو قائمة الانتظار default ): import django_rq
worker = django_rq . get_worker () # Returns a worker for "default" queue
worker . work ()
worker = django_rq . get_worker ( 'low' , 'high' ) # Returns a worker for "low" and "high" لتحويل مهمة قابلة للاستدعاء بسهولة إلى مهمة RQ، يمكنك أيضًا استخدام @job Decorator الذي يأتي مع django_rq :
from django_rq import job
@ job
def long_running_func ():
pass
long_running_func . delay () # Enqueue function in "default" queue
@ job ( 'high' )
def long_running_func ():
pass
long_running_func . delay () # Enqueue function in "high" queueيمكنك تمرير أي وسيطات يقبلها مصمم وظائف RQ:
@ job ( 'default' , timeout = 3600 )
def long_running_func ():
pass
long_running_func . delay () # Enqueue function with a timeout of 3600 seconds. من الممكن تحديد الإعداد الافتراضي لوسيطة الكلمة الرئيسية لـ result_ttl Decorator عبر الإعداد DEFAULT_RESULT_TTL :
RQ = {
'DEFAULT_RESULT_TTL' : 5000 ,
} باستخدام هذا الإعداد، سيقوم مصمم المهام بتعيين result_ttl إلى 5000 ما لم يتم تحديده صراحةً.
يوفر django_rq أمر إدارة يبدأ عاملًا لكل قائمة انتظار محددة كوسائط:
بيثون Manage.py rqworker ارتفاع افتراضي منخفض
إذا كنت تريد تشغيل rqworker في وضع الاندفاع، فيمكنك تمرير علامة --burst :
بيثون Manage.py rqworker عالي الافتراضي منخفض - انفجار
إذا كنت بحاجة إلى استخدام فئات العمال أو الوظائف أو قوائم الانتظار المخصصة، فمن الأفضل استخدام الإعدادات العامة (راجع فئات قوائم الانتظار المخصصة وفئات الوظائف والعاملين المخصصة). ومع ذلك، من الممكن أيضًا تجاوز هذه الإعدادات باستخدام خيارات سطر الأوامر على النحو التالي.
لاستخدام فئة عاملة مخصصة، يمكنك تمرير علامة --worker-class مع المسار إلى العامل الخاص بك:
بيثون Manage.py rqworker عالي الافتراضي منخفض - فئة العمال "path.to.GeventWorker"
لاستخدام فئة قائمة انتظار مخصصة، يمكنك تمرير علامة --queue-class مع المسار إلى فئة قائمة الانتظار الخاصة بك:
بيثون Manage.py rqworker عالي الافتراضي منخفض - فئة قائمة الانتظار "path.to.CustomQueue"
لاستخدام فئة مهمة مخصصة، قم بتوفير علامة --job-class .
بدءًا من الإصدار 2.10، يتم أيضًا دعم تشغيل مجموعة العمال الخاصة بـ RQ:
بيثون Manage.py rqworker-pool الافتراضي منخفض متوسط --num-workers 4
مع آر كيو 1.2.0. يمكنك استخدام برنامج جدولة مدمج لوظائفك. على سبيل المثال:
from django_rq . queues import get_queue
queue = get_queue ( 'default' )
job = queue . enqueue_at ( datetime ( 2020 , 10 , 10 ), func )إذا كنت تستخدم برنامج جدولة مدمج، فيجب عليك بدء العمل بدعم جدولة:
بايثون Manage.py rqworker --with-Scheduler
بدلا من ذلك يمكنك استخدام جدولة RQ. بعد التثبيت، يمكنك أيضًا استخدام الدالة get_scheduler لإرجاع مثيل Scheduler لقوائم الانتظار المحددة في RQ_QUEUES الخاص بـ settings.py. على سبيل المثال:
import django_rq
scheduler = django_rq . get_scheduler ( 'default' )
job = scheduler . enqueue_at ( datetime ( 2020 , 10 , 10 ), func ) يمكنك أيضًا استخدام أمر الإدارة rqscheduler لبدء المجدول:
بيثون Manage.py rqscheduler
إذا كان لديك django-redis أو django-redis-cache مثبتًا، فيمكنك توجيه django_rq لاستخدام نفس معلومات الاتصال من ذاكرة التخزين المؤقت لـ Redis. هذا له ميزتان: إنه جاف ويستفيد من أي تحسين قد يحدث في إعداد ذاكرة التخزين المؤقت (مثل استخدام تجمع الاتصالات أو Hiredis.)
لاستخدام تكوينه، استخدم إملاء مع المفتاح USE_REDIS_CACHE الذي يشير إلى اسم ذاكرة التخزين المؤقت المطلوبة في إملاء RQ_QUEUES الخاص بك. وغني عن القول أن ذاكرة التخزين المؤقت المختارة يجب أن تكون موجودة وتستخدم الواجهة الخلفية لـ Redis. راجع مستندات حزمة Redis ذاكرة التخزين المؤقت الخاصة بك للحصول على تعليمات التكوين. من المهم أيضًا الإشارة إلى أنه نظرًا لأن Django-redis-cache ShardedClient يقسم ذاكرة التخزين المؤقت عبر اتصالات Redis متعددة، فإنه لا يعمل.
فيما يلي مثال لجزء إعدادات Django-redis:
CACHES = {
'redis-cache' : {
'BACKEND' : 'redis_cache.cache.RedisCache' ,
'LOCATION' : 'localhost:6379:1' ,
'OPTIONS' : {
'CLIENT_CLASS' : 'django_redis.client.DefaultClient' ,
'MAX_ENTRIES' : 5000 ,
},
},
}
RQ_QUEUES = {
'high' : {
'USE_REDIS_CACHE' : 'redis-cache' ,
},
'low' : {
'USE_REDIS_CACHE' : 'redis-cache' ,
},
}في بعض الأحيان قد ترغب في تعليق RQ لمنعه من معالجة المهام الجديدة. المثال الكلاسيكي هو أثناء المرحلة الأولية لبرنامج النشر النصي أو قبل وضع موقعك في وضع الصيانة. يعد هذا مفيدًا بشكل خاص عندما تكون لديك وظائف طويلة الأمد نسبيًا وقد يتم قتلها قسراً أثناء النشر.
يمنع أمر الإيقاف المرحلي العاملين في قوائم الانتظار _all_ (في قاعدة بيانات Redis واحدة) من التقاط وظائف جديدة. ومع ذلك، ستستمر المهام الجارية حاليًا حتى اكتمالها.
# Suspend indefinitely
python manage.py rqsuspend
# Suspend for a specific duration (in seconds) then automatically
# resume work again.
python manage.py rqsuspend -d 600
# Resume work again.
python manage.py rqresume يوفر django_rq أيضًا لوحة معلومات لمراقبة حالة قوائم الانتظار الخاصة بك على /django-rq/ (أو أي عنوان URL قمت بتعيينه في urls.py أثناء التثبيت.
يمكنك أيضًا إضافة رابط إلى رابط لوحة المعلومات هذا في /admin عن طريق إضافة RQ_SHOW_ADMIN_LINK = True في settings.py . لكن كن حذرًا، فهذا سيتجاوز قالب المسؤول الافتراضي، لذا قد يتداخل مع التطبيقات الأخرى التي تعدل قالب المسؤول الافتراضي.
تتوفر هذه الإحصائيات أيضًا بتنسيق JSON عبر /django-rq/stats.json ، والذي يمكن للموظفين الوصول إليه. إذا كنت بحاجة إلى الوصول إلى هذا العرض عبر عملاء HTTP آخرين (لأغراض المراقبة)، فيمكنك تحديد RQ_API_TOKEN والوصول إليه عبر /django-rq/stats.json/<API_TOKEN> .
ملاحظة: تعرض إحصائيات المهام المجدولة المهام من برنامج جدولة RQ المدمج، وليس برنامج جدولة RQ الاختياري.
بالإضافة إلى ذلك، يمكن أيضًا الوصول إلى هذه الإحصائيات من سطر الأوامر.
python manage.py rqstats
python manage.py rqstats --interval=1 # Refreshes every second
python manage.py rqstats --json # Output as JSON
python manage.py rqstats --yaml # Output as YAML 
يجب تكوين Sentry ضمن settings.py Django.py كما هو موضح في مستندات Sentry.
يمكنك تجاوز تكوين Django Sentry الافتراضي عند تشغيل الأمر rqworker عن طريق تمرير خيار sentry-dsn :
./manage.py rqworker --sentry-dsn=https://*****@sentry.io/222222
سيؤدي هذا إلى تجاوز أي تكوين موجود لـ Django وإعادة تهيئة Sentry، مع ضبط خيارات Sentry التالية:
{
'debug' : options . get ( 'sentry_debug' ),
'ca_certs' : options . get ( 'sentry_ca_certs' ),
'integrations' : [ RedisIntegration (), RqIntegration (), DjangoIntegration ()]
} يستخدم RQ logging Python، وهذا يعني أنه يمكنك بسهولة تكوين آلية تسجيل rqworker في settings.py الخاص بـ django. على سبيل المثال:
LOGGING = {
"version" : 1 ,
"disable_existing_loggers" : False ,
"formatters" : {
"rq_console" : {
"format" : "%(asctime)s %(message)s" ,
"datefmt" : "%H:%M:%S" ,
},
},
"handlers" : {
"rq_console" : {
"level" : "DEBUG" ,
"class" : "rq.logutils.ColorizingStreamHandler" ,
"formatter" : "rq_console" ,
"exclude" : [ "%(asctime)s" ],
},
},
'loggers' : {
"rq.worker" : {
"handlers" : [ "rq_console" , "sentry" ],
"level" : "DEBUG"
},
}
} بشكل افتراضي، ستستخدم كل قائمة انتظار فئة DjangoRQ . إذا كنت تريد استخدام فئة قائمة انتظار مخصصة، فيمكنك القيام بذلك عن طريق إضافة خيار QUEUE_CLASS على أساس كل قائمة انتظار في RQ_QUEUES :
RQ_QUEUES = {
'default' : {
'HOST' : 'localhost' ,
'PORT' : 6379 ,
'DB' : 0 ,
'QUEUE_CLASS' : 'module.path.CustomClass' ,
}
} أو يمكنك تحديد DjangoRQ لاستخدام فئة مخصصة لجميع قوائم الانتظار الخاصة بك في إعدادات RQ :
RQ = {
'QUEUE_CLASS' : 'module.path.CustomClass' ,
} يجب أن ترث فئات قائمة الانتظار المخصصة من django_rq.queues.DjangoRQ .
إذا كنت تستخدم أكثر من فئة قائمة انتظار واحدة (غير مستحسن)، فتأكد من تشغيل العاملين في قوائم الانتظار التي لها نفس فئة قائمة الانتظار فقط. على سبيل المثال، إذا كان لديك قائمتي انتظار محددتين في RQ_QUEUES وواحدة تحتوي على فئة مخصصة محددة، فسيتعين عليك تشغيل عاملين منفصلين على الأقل لكل قائمة انتظار.
على غرار فئات قائمة الانتظار المخصصة، يمكن تكوين فئات المهام والعاملين المخصصة العامة باستخدام إعدادات JOB_CLASS و WORKER_CLASS :
RQ = {
'JOB_CLASS' : 'module.path.CustomJobClass' ,
'WORKER_CLASS' : 'module.path.CustomWorkerClass' ,
} يجب أن ترث فئة الوظيفة المخصصة من rq.job.Job . سيتم استخدامه لجميع المهام إذا تم تكوينه.
يجب أن ترث الطبقة العاملة المخصصة من rq.worker.Worker . سيتم استخدامه لتشغيل جميع العمال ما لم يتم تجاوزه بواسطة خيار worker-class لأمر إدارة rqworker .
لتسهيل عملية الاختبار، يمكنك تشغيل العامل بشكل متزامن بهذه الطريقة:
from django . test import TestCase
from django_rq import get_worker
class MyTest ( TestCase ):
def test_something_that_creates_jobs ( self ):
... # Stuff that init jobs.
get_worker (). work ( burst = True ) # Processes all jobs then stop.
... # Asserts that the job stuff is done. يمكنك ضبط خيار ASYNC على False لجعل العملية المتزامنة هي العملية الافتراضية لقائمة انتظار معينة. سيؤدي هذا إلى تنفيذ المهام على الفور وعلى نفس مؤشر الترابط الذي تم إرسالها فيه، وهو أمر مفيد للاختبار وتصحيح الأخطاء. على سبيل المثال، يمكنك إضافة ما يلي بعد وضع التكوين في قائمة الانتظار في ملف الإعدادات الخاص بك:
# ... Logic to set DEBUG and TESTING settings to True or False ...
# ... Regular RQ_QUEUES setup code ...
if DEBUG or TESTING :
for queueConfig in RQ_QUEUES . values ():
queueConfig [ 'ASYNC' ] = False لاحظ أن تعيين المعلمة is_async بشكل صريح عند استدعاء get_queue سوف يتجاوز هذا الإعداد.
لتشغيل مجموعة اختبار django_rq :
`` أي django-admin` يختبر django_rq --settings=django_rq.tests.settings --pythonpath=.
قم بإنشاء خدمة rqworker التي تقوم بتشغيل قوائم الانتظار العالية والافتراضية والمنخفضة.
Sudo vi /etc/systemd/system/rqworker.service
[Unit]
Description=Django-RQ Worker
After=network.target
[Service]
WorkingDirectory= << path_to_your_project_folder>>
ExecStart=/home/ubuntu/.virtualenv/<<your_virtualenv>>/bin/python
<<path_to_your_project_folder>>/manage.py
rqworker high default low
[Install]
WantedBy=multi-user.targetتمكين وبدء الخدمة
sudo systemctl enable rqworker
sudo systemctl start rqworkerأضف Django-rq إلى ملف require.txt الخاص بك باستخدام:
pip freeze > requirements.txtقم بتحديث ملفك الشخصي إلى:
web: gunicorn --pythonpath= " $PWD /your_app_name " config.wsgi:application
worker: python your_app_name/manage.py rqworker high default lowالالتزام وإعادة النشر. ثم قم بإضافة العامل الجديد الخاص بك مع:
heroku scale worker=1راجع CHANGELOG.md.
