Django消息队列之django-rq（django 任务队列）

Django消息队列之django-rq（django 任务队列）

github:​​Queue），人如其名，用 redis 做的队列任务

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

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

[安装]

pip install django-rq

添加配置:

修改配置setting.pyINSTALLED_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.0urlpatterns += [ url(r'^django-rq/', include('django_rq.urls')),]# For Django >= 2.0urlpatterns += [ 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(): passlong_running_func.delay() # Enqueue function with a timeout of 3600 seconds.

实战:

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

from django_rq import jobimport logginglogger = logging.getLogger('worker')@job('default', timeout=360)def sync_migration_record(msg): """同步迁移记录""" logger.info(msg)

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

from xxx.tasks import sync_migration_recorddef 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 也有命令可以启动 和查看状态，大家可以自己看看

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。