django rq
v3.0
Redis 기반 Python 대기열 라이브러리인 RQ와 Django 통합. Django-RQ는 django의 settings.py 에서 대기열을 구성하고 프로젝트에서 쉽게 사용할 수 있게 해주는 간단한 앱입니다.
django-rq 유용하다고 생각되면 Tidelift를 통해 개발을 지원하는 것을 고려해 보세요.
django-rq 설치합니다(또는 PyPI에서 다운로드): pip install django - rqsettings.py 의 INSTALLED_APPS 에 django_rq 추가합니다. INSTALLED_APPS = (
# other apps
"django_rq" ,
)settings.py 에서 대기열을 구성합니다. 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 handlersurls.py 에 django_rq.urls 포함하세요. 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 - 선택적 대기열 이름을 허용하고 지정된 대기열(또는 default 대기열)에 대한 새 RQ Worker 인스턴스를 반환합니다. 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 작업으로 쉽게 전환하려면 django_rq 와 함께 제공되는 @job 데코레이터를 사용할 수도 있습니다.
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" queueRQ의 작업 데코레이터가 허용하는 모든 인수를 전달할 수 있습니다.
@ job ( 'default' , timeout = 3600 )
def long_running_func ():
pass
long_running_func . delay () # Enqueue function with a timeout of 3600 seconds. DEFAULT_RESULT_TTL 설정을 통해 result_ttl 데코레이터 키워드 인수에 대한 기본값을 지정할 수 있습니다.
RQ = {
'DEFAULT_RESULT_TTL' : 5000 ,
} 이 설정을 사용하면 작업 데코레이터는 명시적으로 지정하지 않는 한 result_ttl 5000으로 설정합니다.
django_rq는 인수로 지정된 모든 대기열에 대해 작업자를 시작하는 관리 명령을 제공합니다.
python Manage.py rqworker 높음 기본값 낮음
버스트 모드에서 rqworker 실행하려면 --burst 플래그를 전달할 수 있습니다.
python Manage.py rqworker 높음 기본값 낮음 --burst
사용자 정의 작업자, 작업 또는 대기열 클래스를 사용해야 하는 경우 전역 설정을 사용하는 것이 가장 좋습니다(사용자 정의 대기열 클래스 및 사용자 정의 작업 및 작업자 클래스 참조). 그러나 다음과 같이 명령줄 옵션을 사용하여 이러한 설정을 재정의할 수도 있습니다.
사용자 정의 작업자 클래스를 사용하려면 작업자 경로와 함께 --worker-class 플래그를 전달할 수 있습니다.
python 관리.py rqworker 높음 기본값 낮음 --worker-class 'path.to.GeventWorker'
사용자 정의 대기열 클래스를 사용하려면 대기열 클래스 경로와 함께 --queue-class 플래그를 전달할 수 있습니다.
python prepare.py rqworker high default low --queue-class 'path.to.CustomQueue'
사용자 정의 작업 클래스를 사용하려면 --job-class 플래그를 제공하십시오.
버전 2.10부터 RQ의 작업자 풀 실행도 지원됩니다.
python 관리.py rqworker-pool 기본값 낮음 중간 --num-workers 4
RQ 1.2.0. 작업에 내장된 스케줄러를 사용할 수 있습니다. 예를 들어:
from django_rq . queues import get_queue
queue = get_queue ( 'default' )
job = queue . enqueue_at ( datetime ( 2020 , 10 , 10 ), func )내장 스케줄러를 사용하는 경우 스케줄러 지원을 통해 작업자를 시작해야 합니다.
python prepare.py rqworker --with-scheduler
또는 RQ 스케줄러를 사용할 수도 있습니다. 설치 후 get_scheduler 함수를 사용하여 settings.py의 RQ_QUEUES 에 정의된 대기열에 대한 Scheduler 인스턴스를 반환할 수도 있습니다. 예를 들어:
import django_rq
scheduler = django_rq . get_scheduler ( 'default' )
job = scheduler . enqueue_at ( datetime ( 2020 , 10 , 10 ), func ) 관리 명령 rqscheduler 사용하여 스케줄러를 시작할 수도 있습니다.
파이썬 관리.py rqscheduler
django-redis 또는 django-redis-cache가 설치된 경우 django_rq에 Redis 캐시의 동일한 연결 정보를 사용하도록 지시할 수 있습니다. 여기에는 두 가지 장점이 있습니다. DRY이며 캐시 설정에서 진행될 수 있는 최적화(예: 연결 풀링 또는 Hiredis 사용)를 활용합니다.
구성을 사용하려면 RQ_QUEUES 사전에서 원하는 캐시 이름을 가리키는 USE_REDIS_CACHE 키가 있는 사전을 사용하십시오. 선택한 캐시가 존재해야 하고 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를 일시중단할 수도 있습니다. 전형적인 예는 배포 스크립트의 초기 단계 또는 사이트를 유지 관리 모드로 전환하기 전입니다. 이는 상대적으로 오래 실행되고 배포 중에 강제로 종료될 수 있는 작업이 있는 경우 특히 유용합니다.
일시 중지 명령은 _모든_ 대기열(단일 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/ (또는 설치 중에 urls.py 에 설정한 URL)에서 대기열 상태를 모니터링할 수 있는 대시보드도 제공합니다.
settings.py 에 RQ_SHOW_ADMIN_LINK = True 추가하여 /admin 에 이 대시보드 링크에 대한 링크를 추가할 수도 있습니다. 하지만 이는 기본 관리 템플릿을 재정의하므로 기본 관리 템플릿을 수정하는 다른 앱을 방해할 수 있으므로 주의하세요.
이러한 통계는 직원이 액세스할 수 있는 /django-rq/stats.json 을 통해 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 문서에 설명된 대로 Django settings.py 내에서 Sentry를 구성해야 합니다.
sentry-dsn 옵션을 전달하여 rqworker 명령을 실행할 때 기본 Django Sentry 구성을 재정의할 수 있습니다.
./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는 Python의 logging 사용합니다. 이는 django의 settings.py 에서 rqworker 의 로깅 메커니즘을 쉽게 구성할 수 있음을 의미합니다. 예를 들어:
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 클래스를 사용합니다. 사용자 정의 대기열 클래스를 사용하려면 RQ_QUEUES 에 대기열별로 QUEUE_CLASS 옵션을 추가하면 됩니다.
RQ_QUEUES = {
'default' : {
'HOST' : 'localhost' ,
'PORT' : 6379 ,
'DB' : 0 ,
'QUEUE_CLASS' : 'module.path.CustomClass' ,
}
} 또는 RQ 설정에서 모든 대기열에 대해 사용자 정의 클래스를 사용하도록 DjangoRQ 지정할 수 있습니다.
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 에서 상속되어야 합니다. rqworker 관리 명령 worker-class 옵션으로 재정의되지 않는 한 모든 작업자를 실행하는 데 사용됩니다.
더 쉬운 테스트 프로세스를 위해 다음과 같은 방법으로 작업자를 동기적으로 실행할 수 있습니다.
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 get_queue 호출할 때 is_async 매개변수를 명시적으로 설정하면 이 설정이 무시됩니다.
django_rq 의 테스트 스위트를 실행하려면:
` which 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다음을 사용하여 요구 사항.txt 파일에 django-rq를 추가합니다.
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=1CHANGELOG.md를 참조하세요.
