django celery异步框架

描述：实现运维平台的异步执行与定时任务，以下简单描述了安装过程及使用。

安装django和celery

pip install django
pip install celery 
pip install django-celery

新建一个项目名为news

root@bogon:~# django-admin startproject news

查看目录树

root@bogon:~# tree news/
news/
├── manage.py
└── news
├── __init__.py
├── settings.py
├── urls.py
└── wsgi.py

定义一个celery实例 news/news/celery.py

from __future__ import absolute_import
 
import os
 
from celery import Celery
 
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'news.settings')
 
from django.conf import settings
 
app = Celery('news')
 
# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
 
 
@app.task(bind=True)
def debug_task(self):
  print('Request: {0!r}'.format(self.request))

在django中加载应用news/news/__init__.py

from __future__ import absolute_import
 
# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

celery基本配置

import djcelery
djcelery.setup_loader()
 
BROKER_URL = 'django://'
# 以下为mq配置
# BROKER_URL = "amqp://guest:guest@localhost:5672//"
 
INSTALLED_APPS = (
  'djcelery',
  'kombu.transport.django',
)

celery队列配置以及详细说明

# 以下是标准的redis配置
BROKER_URL = 'redis://localhost:6379'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'
 
 
# rabbitmq 配置
# 官网优化的地方也推荐使用c的librabbitmq
CELERY_RESULT_BACKEND = "amqp"
# celery任务执行结果的超时时间
CELERY_TASK_RESULT_EXPIRES = 1200
# celery worker的并发数
CELERYD_CONCURRENCY = 50
# celery worker 每次去rabbitmq取任务的数量
CELERYD_PREFETCH_MULTIPLIER = 4
# 每个worker执行了多少任务就会死掉
CELERYD_MAX_TASKS_PER_CHILD = 40
# 这是使用了django-celery默认的数据库调度模型,任务执行周期都被存在你指定的orm数据库中
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
# 我做的rabbitmq和celery监控很完善所以这个任务超时时间比较短只有半小时
CELERYD_TASK_TIME_LIMIT = 1800
# 默认的队列，如果一个消息不符合其他的队列就会放在默认队列里面
CELERY_DEFAULT_QUEUE = "default_dongwm"
 
## 以下是队列的一些配置
CELERY_QUEUES = {
  "default_dongwm": { # 这是上面指定的默认队列
    "exchange": "default_dongwm",
    "exchange_type": "direct",
    "routing_key": "default_dongwm"
  },
  "topicqueue": { # 这是一个topic队列 凡是topictest开头的routing key都会被放到这个队列
    "routing_key": "topictest.#",
    "exchange": "topic_exchange",
    "exchange_type": "topic",
  },
  "test2": { # test和test2是2个fanout队列,注意他们的exchange相同
    "exchange": "broadcast_tasks",
    "exchange_type": "fanout",
    "binding_key": "broadcast_tasks",
  },
  "test": {
    "exchange": "broadcast_tasks",
    "exchange_type": "fanout",
    "binding_key": "broadcast_tasks2",
  },
}

class MyRouter(object):
  def route_for_task(self, task, args=None, kwargs=None):
    if task.startswith('topictest'):
      return {
        'queue': 'topicqueue',
        }
    # 我的dongwm.tasks文件里面有2个任务都是test开头
    elif task.startswith('webui.tasks.test'):
      return {
        "exchange": "broadcast_tasks",
      }
    # 剩下的其实就会被放到默认队列
    else:
      return None
# CELERY_ROUTES本来也可以用一个大的含有多个字典的字典,但是不如直接对它做一个名称统配
CELERY_ROUTES = (MyRouter(), )

创建一个应用和异步任务

root@bogon:~# django-admin startapp webui
root@bogon:~# cat webui/tasks.py
from __future__ import absolute_import
 
from celery import shared_task
 
@shared_task
def add(x, y):
  return x + y

创建一个定时任务

#每分钟执行任务
@periodic_task(run_every=crontab(minute="*/1"))
def check_ping():
  print 'Pong'

更多详细的配置请参考官方文档。

http://docs.celeryproject.org/en/latest/index.html