Django使用Celery异步任务队列的使用

1 Celery简介

Celery是异步任务队列,可以独立于主进程运行,在主进程退出后,也不影响队列中的任务执行。

任务执行异常退出,重新启动后,会继续执行队列中的其他任务,同时可以缓存停止期间接收的工作任务,这个功能依赖于消息队列(MQ、Redis)。

1.1 Celery原理

Celery的 架构 由三部分组成,消息中间件(message broker),任务执行单元(worker)和任务执行结果存储(task result store)组成。

消息中间件:Celery本身不提供消息服务,但是可以方便的和第三方提供的消息中间件集成。包括, RabbitMQRedis ,  MongoDB  (experimental), Amazon SQS (experimental),CouchDB (experimental), SQLAlchemy (experimental),Django ORM (experimental), IronMQ。推荐使用:RabbitMQ、Redis作为消息队列。

任务执行单元:Worker是Celery提供的任务执行的单元,worker并发的运行在分布式的系统节点中。

任务结果存储:Task result store用来存储Worker执行的任务的结果,Celery支持以不同方式存储任务的结果,包括AMQP, Redis,memcached, MongoDB,SQLAlchemy, Django ORM,Apache Cassandra, IronCache

1.2Celery适用场景

异步任务处理:例如给注册用户发送短消息或者确认邮件任务。 大型任务:执行时间较长的任务,例如视频和图片处理,添加水印和转码等,需要执行任务时间长。 定时执行的任务:支持任务的定时执行和设定时间执行。例如性能压测定时执行。

2Celery开发环境准备

2.1 环境准备


软件名称


版本号


说明


Linux


Centos 6.5(64bit)


操作系统


Python


3.5.2


Django


1.10


Web框架


Celery


4.0.2


异步任务队列


Redis


2.4


消息队列

2.2     Celery安装

使用方法介绍:

Celery的运行依赖消息队列,使用时需要安装redis或者rabbit。

这里我们使用Redis。安装redis库:

sudo yum install redis

启动redis:

sudo service redis start

安装celery库

sudo pip install celery==4.0.2

3Celery单独执行任务

3.1编写任务

创建task.py文件

说明:这里初始Celery实例时就加载了配置,使用的redis作为消息队列和存储任务结果。

运行celery:

$ celery -A task worker --loglevel=info

看到下面的打印,说明celery成功运行。

3.2 调用任务

直接打开python交互命令行

执行下面代码:

可以celery的窗口看到任务的执行信息

任务执行状态监控和获取结果:

3.3任务调用方法总结

有两种方法:

delay和apply_async ,delay方法是apply_async简化版。

add.delay(2, 2)
add.apply_async((2, 2))
add.apply_async((2, 2), queue='lopri')

delay方法是apply_async简化版本。

apply_async方法是可以带非常多的配置参数,包括指定队列等

Queue 指定队列名称,可以把不同任务分配到不同的队列 3.4     任务状态

每个任务有三种状态:PENDING -> STARTED -> SUCCESS

任务查询状态:res.state

来查询任务的状态

4与Django集成

上面简单介绍了celery异步任务的基本方法,结合我们实际的应用,我们需要与Django一起使用,下面介绍如何与Django结合。

4.1与Django集成方法

与Django集成有两种方法:

  1. Django 1.8 以上版本:与Celery 4.0版本集成
  2. Django 1.8 以下版本:与Celery3.1版本集成,使用django-celery库

今天我们介绍celery4.0 和django 1.8以上版本集成方法。

4.2 创建项目文件

创建一个项目:名字叫做proj

- proj/
 - proj/__init__.py
 - proj/settings.py
 - proj/urls.py
 - proj/wsgi.py
- manage.py

创建一个新的文件: proj/proj/mycelery.py

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

app = Celery('proj')

# Using a string here means the worker don't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
# should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

在proj/proj/__init__.py:添加

from __future__ import absolute_import, unicode_literals

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

__all__ = ['celery_app']

4.3 配置Celery

我们在mycelery.py文件中说明celery的配置文件在settings.py中,并且是以CELERY开头。

app.config_from_object('django.conf:settings', namespace='CELERY')

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

我们的配置是使用redis作为消息队列,消息的代理和结果都是用redis,任务的序列化使用json格式。

重要:redis://127.0.0.1:6379/0这个说明使用的redis的0号队列,如果有多个celery任务都使用同一个队列,则会造成任务混乱。最好是celery实例单独使用一个队列。

4.4创建APP

创建Django的App,名称为celery_task,在app目录下创建tasks.py文件。

完成后目录结构为:

├── celery_task
│ ├── admin.py
│ ├── apps.py
│ ├── __init__.py
│ ├── migrations
│ │ └── __init__.py
│ ├── models.py
│ ├── tasks.py
│ ├── tests.py
│ └── views.py
├── db.sqlite3
├── manage.py
├── proj
│ ├── celery.py
│ ├── __init__.py
│ ├── settings.py
│ ├── urls.py
│ └── wsgi.py
└── templates

4.5编写task任务

编辑任务文件

tasks.py

在tasks.py文件中添加下面代码

# Create your tasks here
from __future__ import absolute_import, unicode_literals
from celery import shared_task

@shared_task
def add(x, y):
 return x + y

@shared_task
def mul(x, y):
 return x * y

@shared_task
def xsum(numbers):
 return sum(numbers)

启动celery:celery -A proj.mycelery worker -l info

说明:proj 为模块名称,mycelery 为celery 的实例所在的文件。

启动成功打印:

4.6在views中调用任务

在views中编写接口,实现两个功能:

  1. 触发任务,然后返回任务的结果和任务ID
  2. 根据任务ID查询任务状态

代码如下:

启动django。

新开一个会话启动celery;启动命令为:celery –A proj.mycelery worker –l info

访问 http://127.0.0.1:8000/add ,可以看到返回的结果。

在celery运行的页面,可以看到下面输出:

4.7在views中查询任务状态

有的时候任务执行时间较长,需要查询任务是否执行完成,可以根据任务的id来查询任务状态,根据状态进行下一步操作。

可以看到任务的状态为:SUCCESS

5Celery定时任务

Celery作为异步任务队列,我们可以按照我们设置的时间,定时的执行一些任务,例如每日数据库备份,日志转存等。

Celery的定时任务配置非常简单:

定时任务的配置依然在setting.py文件中。

说明:如果觉得celery 的数据配置文件和Django 的都在setting.py 一个文件中不方便,可以分拆出来,只需要在mycelery.py 的文件中指明即可。

app.config_from_object('django.conf:yoursettingsfile', namespace='CELERY')

5.1任务间隔运行

#每30秒调用task.add
from datetime import timedelta

CELERY_BEAT_SCHEDULE = {
 'add-every-30-seconds': {
  'task': 'tasks.add',
  'schedule': timedelta(seconds=30),
  'args': (16, 16)
 },
}

5.2定时执行

定时每天早上7:30分运行。

注意:设置任务时间时注意时间格式,UTC时间或者本地时间。

#crontab任务
#每天7:30调用task.add
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
 # Executes every Monday morning at 7:30 A.M
 'add-every-monday-morning': {
  'task': 'tasks.add',
  'schedule': crontab(hour=7, minute=30),
  'args': (16, 16),
 },
}

5.3定时任务启动

配置了定时任务,除了worker进程外,还需要启动一个beat进程。

Beat进程的作用就相当于一个定时任务,根据配置来执行对应的任务。

5.3.1  启动beat进程

命令如下:celery -A proj.mycelery beat -l info

5.3.2  启动worker进程

Worker进程启动和前面启动命令一样。celery –A proj.mycelery worker –l info

6 Celery深入

Celery任务支持多样的运行模式:

  1. 支持动态指定并发数 --autoscale=10,3 (always keep 3 processes, but grow to 10 if necessary).
  2. 支持链式任务
  3. 支持Group任务
  4. 支持任务不同优先级
  5. 支持指定任务队列
  6. 支持使用eventlet模式运行worker

例如:指定并发数为1000

celery -A proj.mycelery worker -c 1000

这些可以根据使用的深入自行了解和学习。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

时间: 2018-03-10

django中使用Celery 布式任务队列过程详解

本文记录django中如何使用celery完成异步任务. Celery 是一个简单.灵活且可靠的,处理大量消息的分布式系统,并且提供维护这样一个系统的必需工具. 它是一个专注于实时处理的任务队列,同时也支持任务调度. 官方网站 中文文档 示例一:用户发起request,并等待response返回.在本些views中,可能需要执行一段耗时的程序,那么用户就会等待很长时间,造成不好的用户体验 示例二:网站每小时需要同步一次天气预报信息,但是http是请求触发的,难道要一小时请求一次吗? 使用cele

Django Celery异步任务队列的实现

背景 在开发中,我们常常会遇到一些耗时任务,举个例子: 上传并解析一个 1w 条数据的 Excel 文件,最后持久化至数据库. 在我的程序中,这个任务耗时大约 6s,对于用户来说,6s 的等待已经是个灾难了. 比较好的处理方式是: 接收这个任务的请求 将这个任务添加到队列中 立即返回「操作成功,正在后台处理」的字样 后台消费这个队列,执行这个任务 我们按照这个思路,借助 Celery 进行实现. 实现 本文所使用的环境如下: Python 3.6.7 RabbitMQ 3.8 Celery 4.

异步任务队列Celery在Django中的使用方法

前段时间在Django Web平台开发中,碰到一些请求执行的任务时间较长(几分钟),为了加快用户的响应时间,因此决定采用异步任务的方式在后台执行这些任务.在同事的指引下接触了Celery这个异步任务队列框架,鉴于网上关于Celery和Django结合的文档较少,大部分也只是粗粗介绍了大概的流程,在实践过程中还是遇到了不少坑,希望记录下来帮助有需要的朋友. 一.Django中的异步请求 Django Web中从一个http请求发起,到获得响应返回html页面的流程大致如下:http请求发起 --

Django配置celery(非djcelery)执行异步任务和定时任务

所有演示均基于Django2.0 celery是一个基于python开发的简单.灵活且可靠的分布式任务队列框架,支持使用任务队列的方式在分布式的机器/进程/线程上执行任务调度.采用典型的生产者-消费者模型,主要由三部分组成: 消息队列broker:broker实际上就是一个MQ队列服务,可以使用redis.rabbitmq等作为broker 处理任务的消费者workers:broker通知worker队列中有任务,worker去队列中取出任务执行,每一个worker就是一个进程 存储结果的bac

python使用celery实现异步任务执行的例子

使用celery在django项目中实现异步发送短信 在项目的目录下创建celery_tasks用于保存celery异步任务. 在celery_tasks目录下创建config.py文件,用于保存celery的配置信息 ```broker_url = "redis://127.0.0.1/14"``` 在celery_tasks目录下创建main.py文件,用于作为celery的启动文件 from celery import Celery # 为celery使用django配置文件进行

使用celery执行Django串行异步任务的方法步骤

前言 Django项目有一个耗时较长的update过程,希望在接到请求运行update过程的时候,Django应用仍能正常处理其他的请求,并且update过程要求不能并行,也不能漏掉任何一个请求 使用celery的solo模式解决 安装redis https://github.com/microsoftarchive/redis/releases 下载.msi文件安装,会直接将redis注册为windows服务 安装celery与redis依赖 pip install celery pip in

Django中使用celery完成异步任务的示例代码

本文主要介绍如何在django中用celery完成异步任务,web项目中为了提高用户体验可以对一些耗时操作放到异步队列中去执行,例如激活邮件,后台计算操作等等 当前项目环境为: django==1.11.8 celery==3.1.25 redis==2.10.6 pip==9.0.1 python3==3.5.2 django-celery==3.1.17 一,创建Django项目及celery配置 1,创建Django项目 1>打开终端输入:django-admin startproject

python中利用Future对象异步返回结果示例代码

前言 本文主要给大家介绍了关于python中用Future对象异步返回结果的相关内容,分享出来供大家参考学习,下面话不多说了,来一起看看详细的介绍吧. 一个Future是用来表示将来要完成的结果,异步循环可以自动完成对这种对象的状态触发. 例子如下: import asyncio def mark_done(future, result): print('setting future result to {!r}'.format(result)) future.set_result(result

Django中使用Celery的教程详解

Django教程 Python下有许多款不同的 Web 框架.Django是重量级选手中最有代表性的一位.许多成功的网站和APP都基于Django. Django是一个开放源代码的Web应用框架,由Python写成. Django遵守BSD版权,初次发布于2005年7月, 并于2008年9月发布了第一个正式版本1.0 . Django采用了MVC的软件设计模式,即模型M,视图V和控制器C. 一.前言 Celery是一个基于python开发的分布式任务队列,如果不了解请阅读笔者上一篇博文Celer

Django中使用Celery的方法示例

起步 在 <分布式任务队列Celery使用说明> 中介绍了在 Python 中使用 Celery 来实验异步任务和定时任务功能.本文介绍如何在 Django 中使用 Celery. 安装 pip install django-celery 这个命令使用的依赖是 Celery 3.x 的版本,所以会把我之前安装的 4.x 卸载,不过对功能上并没有什么影响.我们也完全可以仅用Celery在django中使用,但使用 django-celery 模块能更好的管理 celery. 使用 可以把有关 C

django中上传图片分页三级联动效果的实现代码

Django1.8.2中文文档:Django1.8.2中文文档 上传图片配置上传文件保存目录 1)新建上传文件保存目录. 2)配置上传文件保存目录. 后台管理页面上传图片 1)设计模型类. 2)迁移生成表格. 3) 注册模型类. 后台管理页面上传图片实例 1.在static下面创建media文件夹(再在media文件夹里面新建booktest文件夹). 2.设置静态文件保存目录 # 设置上传文件的保存目录 MEDIA_ROOT = os.path.join(BASE_DIR, 'static/m

Django中的模型类设计及展示示例详解

django中设计数据模型类是基于ORM的对象关系映射更方便的进行数据库中的数据操作. 对象关系映射 把面向对象中的类和数据库表--对应,通过操作类和对象,对数表实现数据操作,不需要写sql,由ORM框架生成 django实现了ORM框架,在项目中与数据库之间产生桥梁作用 django数据库定义模型的步骤如下: python manage.py makemigrations python mange.py migrate 在应用models.py中编写模型类,继承models.Model类 在模

Django Channel实时推送与聊天的示例代码

先来看一下最终的效果吧 开始聊天,输入消息并点击发送消息就可以开始聊天了 点击 "获取后端数据"开启实时推送 先来简单了解一下 Django Channel Channels是一个采用Django并将其功能扩展到HTTP以外的项目,以处理WebSocket,聊天协议,IoT协议等.它基于称为ASGI的Python规范构建. 它以Django的核心为基础,并在其下面分层了一个完全异步的层,以同步模式运行Django本身,但异步处理了连接和套接字,并提供了以两种方式编写的选择,从而实现了这

django在保存图像的同时压缩图像示例代码详解

假设我们有一个非常简单的Post模型,它将是一个图像及其描述, from django.db import models class Post(models.Model): text = models.TextField() image = models.ImageField(upload_to='images/') 但是我们要优化图像大小,这将由我们Post的image字段指出. 这样做有充分的理由-它有助于更快地加载网站/应用程序并减少我们的服务器存储. 在使用Django之前,首先让我们简

nodejs中使用HTTP分块响应和定时器示例代码

在本例中,将要创建一个输出纯文本的HTTP服务器,输出的纯文本每隔一秒会新增100个用换行符分隔的时间戳. require('http').createServer(function(req, res) { res.writeHead(200, {'Content-Type': 'text/plain'}); var left = 10; var interval = setInterval(function() { for(var i = 0; i< 100; i++) { res.write

python 中if else 语句的作用及示例代码

引入:if-else的作用,满足一个条件做什么,否则做什么. if-else语句语法结构 if 判断条件: 要执行的代码 else: 要执行的代码 判断条件:一般为关系表达式或bool类型的值 执行过程:程序运行到if处,首先判断所带的条件,如果条件成立,就是返回值是True,则执行下面的代码:如果条件不成立则返回值是False, 则继续执行下面的代码. 示例1:模拟用户登录 提示输入用户名和密码 如果用户名是Admin,密码等于123.com, 提示用户登录成功 如果用户名不是Admin,提示