Django消息队列之django-rq

github:​​https://github.com/rq/django-rq​​

 

RQ（Redis Queue），人如其名，用 redis 做的队列任务

redis ，众所周知， 它的列表可以做队列，rq就是把job放进队列里，然后启worker挨个做完

另外rq极其简单，官方文档短小精悍，容易上手

[安装]

pip install django-rq

添加配置:

修改配置setting.py

INSTALLED_APPS = [
...
"django_rq",
]

注意: 这里使用的下划线,
RQ_QUEUES = {
'default': {
'HOST': 'localhost',
'PORT': 6379,
'DB': 0,
'PASSWORD': 'some-password',
'DEFAULT_TIMEOUT': 360,
},
'with-sentinel': {
'SENTINELS': [('localhost', 26736), ('localhost', 26737)],
'MASTER_NAME': 'redismaster',
'DB': 0,
'PASSWORD': 'secret',
'SOCKET_TIMEOUT': None,
'CONNECTION_KWARGS': {
'socket_connect_timeout': 0.3
},
},
'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 handlers



#default,high,low表示队列的优先级,high > default > low
# 但是如果一大批low队列的job在执行的话，此时high队列开始入队job,不会马上下一个任务就开始执行high队列，而是会继续执行low队列，直至low队列任务执行完毕

支持使用django-redis和django-redis-cache

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',
},
}

添加日志配置:

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.utils.ColorizingStreamHandler",
"formatter": "rq_console",
"exclude": ["%(asctime)s"],
},
# If you use sentry for logging
'sentry': {
'level': 'ERROR',
'class': 'raven.contrib.django.handlers.SentryHandler',
},
},
'loggers': {
"rq.worker": {
"handlers": ["rq_console", "sentry"],
"level": "DEBUG"
},
}
}

添加路由:

修改urls.py

# For Django < 2.0
urlpatterns += [
url(r'^django-rq/', include('django_rq.urls')),
]

# For Django >= 2.0
urlpatterns += [
path('django-rq/', include('django_rq.urls'))
# 添加后台查看
path(r'admin/django-rq/', include('django_rq.urls')),
]

使用:

1.使用@job装饰器

 

@job('default', timeout=3600)
def long_running_func():
pass
long_running_func.delay() # Enqueue function with a timeout of 3600 seconds.

 

实战:

在项目目录下添加tasks.py处理任务文件

from django_rq import job

import logging
logger = logging.getLogger('worker')

@job('default', timeout=360)
def sync_migration_record(msg):
"""同步迁移记录"""
logger.info(msg)

然后在任意一个视图文件中添加任务:

from xxx.tasks import sync_migration_record

def test_task(request):
sync_migration_record.delay("hello,rq")

 

指定任务:

python manage.py rqworker high default low

#在 high default low三个队列各自启动一个worker，注意了，由于django_rq调用linux中fork()，所以只能在linux系统中执行，windos可以尝试win10的子系统

更多:

查看队列执行状况

多种办法

1.配置好的django admin中查看，种类齐全，最佳查看方式

2.python manage.py rqstats

python manage.py rqstats --interval=1  #每秒刷新（其实刷新并不及时）

python manage.py rqstats --json  # 输出JSON

python manage.py rqstats --yaml  # 输出YAML

3. 进入redis库中可以看到自己的队列，worker，以及job(作为辅助验证使用)

4.rq 也有命令可以启动 和查看状态，大家可以自己看看

 

【转自：韩国lg机房 http://www.558idc.com/lg.html欢迎留下您的宝贵建议】