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