Django中使用Celery的方法示例

yipeiwu_com5年前Python基础

起步

在 《分布式任务队列Celery使用说明》 中介绍了在 Python 中使用 Celery 来实验异步任务和定时任务功能。本文介绍如何在 Django 中使用 Celery。

安装

pip install django-celery

这个命令使用的依赖是 Celery 3.x 的版本,所以会把我之前安装的 4.x 卸载,不过对功能上并没有什么影响。我们也完全可以仅用Celery在django中使用,但使用 django-celery 模块能更好的管理 celery。

使用

可以把有关 Celery 的配置放到 settings.py 里去,但我比较习惯单独一个文件来放,然后在 settings.py 引入进来:

# celery_config.py
import djcelery
import os

os.environ.setdefault('FORKED_BY_MULTIPROCESSING', '1')
djcelery.setup_loader()

BROKER_URL = 'redis://127.0.0.1:6379/1'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/2'

# UTC
CELERY_ENABLE_UTC = True
CELERY_TIMEZONE = 'Asia/Shanghai'

CELERY_IMPORTS = (
 'app.tasks',
)

# 有些情况可以防止死锁
CELERY_FORCE_EXECV = True

# 设置并发的worker数量
CELERYD_CONCURRENCY = 4

# 任务发送完成是否需要确认,这一项对性能有一点影响
CELERY_ACKS_LATE = True

# 每个worker执行了多少任务就会销毁,防止内存泄露,默认是无限的
CELERYD_MAX_TASKS_PER_CHILD = 40

# 规定完成任务的时间
CELERYD_TASK_TIME_LIMIT = 15 * 60 # 在15分钟内完成任务,否则执行该任务的worker将被杀死,任务移交给父进程

# 设置默认的队列名称,如果一个消息不符合其他的队列就会放在默认队列里面,如果什么都不设置的话,数据都会发送到默认的队列中
CELERY_DEFAULT_QUEUE = "default"

# 设置详细的队列
CELERY_QUEUES = {
 "default": { # 这是上面指定的默认队列
  "exchange": "default",
  "exchange_type": "direct",
  "routing_key": "default"
 },
 "beat_queue": {
  "exchange": "beat_queue",
  "exchange_type": "direct",
  "routing_key": "beat_queue"
 }

}

配置文件中设置了 CELERY_IMPORTS 导入的任务,所以在django app中创建相应的任务文件:

# app/tasks.py
from celery.task import Task
import time

class TestTask(Task):
 name = 'test-task' # 给任务设置个自定义名称

 def run(self, *args, **kwargs):
  print('start test task')
  time.sleep(4)
  print('args={}, kwargs={}'.format(args, kwargs))
  print('end test task')

在 settings.py 添加:

INSTALLED_APPS = [
 # ...
 'djcelery',
]

# Celery
from learn_django.celery_config import *

触发任务或提交任务可以在view中来调用:

# views.py
from django.http import HttpResponse
from app.tasks import TestTask

def test_task(request):
 # 执行异步任务
 print('start do request')
 t = TestTask()
 t.delay()
 print('end do request')
 return HttpResponse('ok')

启动 woker 的命令是:

python manage.py celery worker -l info

再启动django,访问该view,可以看到任务在worker中被消费了。

定时任务

在celery的配置文件 celery_config.py 文件中添加:

CELERYBEAT_SCHEDULE = {
 'task1-every-1-min': { # 自定义名称
  'task': 'test-task', # 与任务中name名称一致
  'schedule': datetime.timedelta(seconds=5),
  'args': (2, 15),
  'options': {
   'queue': 'beat_queue', # 指定要使用的队列
  }
 },

}

通过 options 的 queque 来指定要使用的队列,这里需要单独的队列是因为,如果所有任务都使用同一队列,对于定时任务来说,任务提交后会位于队列尾部,任务的执行时间会靠后,所以对于定时任务来说,使用单独的队列。

启动 beat:

python manage.py celery beat -l info

监控工具 flower

如果celery中的任务执行失败了,有些场景是需要对这些任务进行监控, flower 是基于 Tornado 开发的web应用。安装用 pip install flower ;启动它可以是:

python manage.py celery flower

# python manage.py celery flower --basic_auth=admin:admin

用浏览器访问 http://localhost:5555 即可查看:

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持【听图阁-专注于Python设计】。

相关文章

python通过文件头判断文件类型

对于提供上传的服务器,需要对上传的文件进行过滤。 本文为大家提供了python通过文件头判断文件类型的方法,避免不必要的麻烦。 分享代码如下 import struct # 支...

python PyQt5/Pyside2 按钮右击菜单实例代码

具体代码如下所述: import sys from PySide2.QtGui import * from PySide2.QtCore import * from PySide2....

对python读取zip压缩文件里面的csv数据实例详解

对python读取zip压缩文件里面的csv数据实例详解

利用zipfile模块和pandas获取数据,代码比较简单,做个记录吧: # -*- coding: utf-8 -*- """ Created on Tue Aug 21 22:3...

python使用adbapi实现MySQL数据库的异步存储

python使用adbapi实现MySQL数据库的异步存储

之前一直在写有关scrapy爬虫的事情,今天我们看看使用scrapy如何把爬到的数据放在MySQL数据库中保存。 有关python操作MySQL数据库的内容,网上已经有很多内容可以参考了...

Python有序查找算法之二分法实例分析

Python有序查找算法之二分法实例分析

本文实例讲述了Python有序查找算法之二分法。分享给大家供大家参考,具体如下: 二分法是一种快速查找的方法,时间复杂度低,逻辑简单易懂,总的来说就是不断的除以2除以2... 例如需要查...