Celery 异步任务、定时任务

时间  2019-11-12 标签 celery 异步 任务 定时

celery介绍

Celery是一个功能完备即插即用的任务队列。它使得咱们不须要考虑复杂的问题，使用很是简单。celery看起来彷佛很庞大，本章节咱们先对其进行简单的了解，而后再去学习其余一些高级特性。 celery适用异步处理问题，当发送邮件、或者文件上传, 图像处理等等一些比较耗时的操做，咱们可将其异步执行，这样用户不须要等待好久，提升用户体验。 celery的特色是：html

  • 简单，易于使用和维护，有丰富的文档。
  • 高效，单个celery进程每分钟能够处理数百万个任务。
  • 灵活，celery中几乎每一个部分均可以自定义扩展。

celery很是易于集成到一些web开发框架中.python

Task Queue

任务队列是一种跨线程、跨机器工做的一种机制.web

任务队列中包含称做任务的工做单元。有专门的工做进程持续不断的监视任务队列，并从中得到新的任务并处理.redis

celery经过消息进行通讯，一般使用一个叫Broker(中间人)来协client(任务的发出者)和worker(任务的处理者). clients发出消息到队列中，broker将队列中的信息派发给worker来处理。sql

一个celery系统能够包含不少的worker和broker，可加强横向扩展性和高可用性能。数据库

 Celery 安装

咱们可使用python的包管理器pip来安装:django

pip install -U Celery

也可从官方直接下载安装包:https://pypi.python.org/pypi/celery/json

tar xvfz celery-0.0.0.tar.gz
cd celery-0.0.0
python setup.py build
python setup.py install

 Broker

Celery须要一种解决消息的发送和接受的方式，咱们把这种用来存储消息的的中间装置叫作message broker, 也可叫作消息中间人。 做为中间人，咱们有几种方案可选择：windows

1.RabbitMQ

RabbitMQ是一个功能完备，稳定的而且易于安装的broker. 它是生产环境中最优的选择。使用RabbitMQ的细节参照如下连接：http://docs.celeryproject.org/en/latest/getting-started/brokers/rabbitmq.html#broker-rabbitmqcentos

若是咱们使用的是Ubuntu或者Debian发行版的Linux，能够直接经过下面的命令安装RabbitMQ: sudo apt-get install rabbitmq-server 安装完毕以后，RabbitMQ-server服务器就已经在后台运行。若是您用的并非Ubuntu或Debian, 能够在如下网址：http://www.rabbitmq.com/download.html 去查找本身所须要的版本软件。

2.Redis

Redis也是一款功能完备的broker可选项，可是其更可能因意外中断或者电源故障致使数据丢失的状况。 关因而有那个Redis做为Broker，可访下面网址： http://docs.celeryproject.org/en/latest/getting-started/brokers/redis.html#broker-redis

Application

  使用celery第一件要作的最为重要的事情是须要先建立一个Celery实例，咱们通常叫作celery应用，或者更简单直接叫作一个app。app应用是咱们使用celery全部功能的入口，好比建立任务，管理任务等，在使用celery的时候，app必须可以被其余的模块导入。

1.建立应用

咱们首先建立tasks.py模块, 其内容为:

from celery import Celery

# 咱们这里案例使用redis做为broker
app = Celery('demo', broker='redis://:332572@127.0.0.1/1')

# 建立任务函数
@app.task
def my_task():
    print("任务函数正在执行....")

  Celery第一个参数是给其设定一个名字， 第二参数咱们设定一个中间人broker, 在这里咱们使用Redis做为中间人。my_task函数是咱们编写的一个任务函数， 经过加上装饰器app.task, 将其注册到broker的队列中。

  如今咱们在建立一个worker， 等待处理队列中的任务.打开终端，cd到tasks.py同级目录中，执行命令:

celery -A tasks worker --loglevel=info

显示效果以下: 

2.调用任务

  任务加入到broker队列中，以便刚才咱们建立的celery workder服务器可以从队列中取出任务并执行。如何将任务函数加入到队列中，可以使用delay()。

进入python终端, 执行以下代码:

from tasks import my_task
my_task.delay()

执行效果以下: 

 

咱们经过worker的控制台，能够看到咱们的任务被worker处理。调用一个任务函数，将会返回一个AsyncResult对象，这个对象能够用来检查任务的状态或者得到任务的返回值。

3.存储结果

  若是咱们想跟踪任务的状态，Celery须要将结果保存到某个地方。有几种保存的方案可选:SQLAlchemy、Django ORM、Memcached、 Redis、RPC (RabbitMQ/AMQP)。

  例子咱们仍然使用Redis做为存储结果的方案，任务结果存储配置咱们经过Celery的backend参数来设定。咱们将tasks模块修改以下:

from celery import Celery

# 咱们这里案例使用redis做为broker
app = Celery('demo',
             backend='redis://:332572@127.0.0.1:6379/2',
             broker='redis://:332572@127.0.0.1:6379/1')

# 建立任务函数
@app.task
def my_task(a, b):
    print("任务函数正在执行....")
    return a + b

  咱们给Celery增长了backend参数，指定redis做为结果存储,并将任务函数修改成两个参数，而且有返回值。 

更多关于result对象信息，请参阅下列网址: http://docs.celeryproject.org/en/latest/reference/celery.result.html#module-celery.result

 配置

  Celery使用简单，配置也很是简单。Celery有不少配置选项可以使得celery可以符合咱们的须要，可是默认的几项配置已经足够应付大多数应用场景了。

  配置信息能够直接在app中设置，或者经过专有的配置模块来配置。

1.直接经过app来配置

from celery import Celery
app = Celery('demo')
# 增长配置
app.conf.update(
    result_backend='redis://:332572@127.0.0.1:6379/2',
    broker_url='redis://:332572@127.0.0.1:6379/1',
)

2.专有配置文件

  对于比较大的项目，咱们建议配置信息做为一个单独的模块。咱们能够经过调用app的函数来告诉Celery使用咱们的配置模块。

  配置模块的名字咱们取名为celeryconfig, 这个名字不是固定的，咱们能够任意取名，建议这么作。咱们必须保证配置模块可以被导入。 

  下面咱们在tasks.py模块 同级目录下建立配置模块celeryconfig.py:

result_backend = 'redis://:332572@127.0.0.1:6379/2'
broker_url = 'redis://:332572@127.0.0.1:6379/1'

  tasks.py文件修改成:

from celery import Celery
import celeryconfig

# 咱们这里案例使用redis做为broker
app = Celery('demo')

# 从单独的配置模块中加载配置
app.config_from_object('celeryconfig')

更多配置: http://docs.celeryproject.org/en/latest/userguide/configuration.html#configuration

 在项目中使用Celery

1. 配置celery

  建立django项目celery_demo, 建立应用demo:

django-admin startproject celery_demo
python manage.py startapp demo

  在celery_demo模块中建立celery.py模块, 文件目录为:

 

   celery.py模块内容为:

from celery import Celery
from django.conf import settings
import os

# 为celery设置环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings')

# 建立应用
app = Celery("demo")
# 配置应用
app.conf.update(
    # 配置broker, 这里咱们用redis做为broker
    BROKER_URL='redis://:332572@127.0.0.1:6379/1',
)
# 设置app自动加载任务
# 从已经安装的app中查找任务
app.autodiscover_tasks(settings.INSTALLED_APPS)

  在应用demo引用建立tasks.py模块, 文件目录为: 

 

  咱们在文件内建立一个任务函数my_task:

from celery_demo.celery import app
import time

# 加上app对象的task装饰器
# 此函数为任务函数
@app.task
def my_task():
    print("任务开始执行....")
    time.sleep(5)
    print("任务执行结束....")

  在views.py模块中建立视图index:

from django.shortcuts import render
from django.http import HttpResponse
from .tasks import my_task


def index(request):
# 将my_task任务加入到celery队列中
# 若是my_task函数有参数，可经过delay()传递
# 例如 my_task(a, b), my_task.delay(10, 20)
    my_task.delay()

    return HttpResponse("<h1>服务器返回响应内容!</h1>")

  在celey_demo/settings.py配置视图路由:

from django.conf.urls import url
from django.contrib import admin
from demo.views import index

urlpatterns = [
    url(r'^admin/', admin.site.urls),
    url(r'^$', index),
]

  建立worker等待处理celery队列中任务, 在终端执行命令:

celery -A celery_demo worker -l info

 

  启动django测试服务器：

python manage.py runserver

 

2. 存储任务结果

  此处须要用到额外包django_celery_results, 先安装包:

pip install django-celery-results

  在celery_demo/settings.py中安装此应用:

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'demo',
    'django_celery_results',  # 注意此处应用名为下划线
]

  回到celery_demo/celery.py模块中，增长配置信息以下:

from celery import Celery
from django.conf import settings
import os

# 为celery设置环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings')

# 建立应用
app = Celery("demo")
# 配置应用
app.conf.update(
    # 配置broker, 这里咱们用redis做为broker
    BROKER_URL='redis://:332572@127.0.0.1:6379/1',
    # 使用项目数据库存储任务执行结果
    CELERY_RESULT_BACKEND='django-db',
)
# 设置app自动加载任务
# 从已经安装的app中查找任务
app.autodiscover_tasks(settings.INSTALLED_APPS)

  建立django_celery_results应用所需数据库表, 执行迁移文件：

python manage.py migrate django_celery_results

  我这里使用的是django默认的数据库sqlit, 执行迁移以后，会在数据库中建立一张用来存储任务结果的表: 

 

3. 定时任务

  若是咱们想某日某时执行某个任务，或者每隔一段时间执行某个任务，也可使用celery来完成.   使用定时任务，须要安装额外包:

pip install django_celery_beat

  首先在settings.py中安装此应用:

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'demo',
    'django_celery_results',
    'django_celery_beat',  # 安装应用
]

  在celery_demo/celery.py模块中增长定时任务配置:

from celery import Celery
from django.conf import settings
import os

# 为celery设置环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings')

# 建立应用
app = Celery("demo")
# 配置应用
app.conf.update(
    # 配置broker, 这里咱们用redis做为broker
    BROKER_URL='redis://:332572@127.0.0.1:6379/1',
    # 使用项目数据库存储任务执行结果
    CELERY_RESULT_BACKEND='django-db',
    # 配置定时器模块，定时器信息存储在数据库中
    CELERYBEAT_SCHEDULER='django_celery_beat.schedulers.DatabaseScheduler',

)
# 设置app自动加载任务
# 从已经安装的app中查找任务
app.autodiscover_tasks(settings.INSTALLED_APPS)

  因为定时器信息存储在数据库中，咱们须要先生成对应表, 对diango_celery_beat执行迁移操做，建立对应表:

python manage.py migrate django_celery_beat

  咱们可登陆网站后台Admin去建立对应任务, 首先咱们先在tasks.py模块中增长新的任务，用于定时去执行(5秒执行一次)

from celery_demo.celery import app
import time

# 用于定时执行的任务
@app.task
def interval_task():
    print("我每隔5秒钟时间执行一次....")

  首先建立后台管理员账号:

python manage.py createsuperuser

  登陆管理后台Admin:

     其中Crontabs用于定时某个具体时间执行某个任务的时间，Intervals用于每隔多久执行任务的事件，具体任务的执行在Periodic tasks表中建立。

  咱们要建立每隔5秒执行某个任务，因此在Intervals表名后面点击Add按钮:

  而后在Periodic tasks表名后面，点击Add按钮，添加任务:

  启动定时任务：

celery -A celery_demo worker -l info --beat

   任务每隔5秒中就会执行一次，若是配置了存储，那么每次任务执行的结果也会被保存到对应的数据库中。

调用tasks.py的方法时，有时会出现传入的参数没法被序列表化，解决方法：将传入的参数以字典的形式包装成一个字典，确保字典可以被json序列化，及key和value都是能被json序列化的。

在centos上使用pip或者编译安装了celery却提示找不到celery这个命令时，解决方法：

export PATH=/usr/local/python3/bin:$PATH      # 个人Python3资源文件都放在 /usr/local/python3 文件夹下

echo 'export PATH=/usr/local/python3/bin:$PATH' >> /etc/profile.d/python3.sh  # 避免重启丢失

若是提示没有权限 就加sudo

在windows10上使用celery，使用celery -A celery_demo worker -l info命令出错,解决方法：安装协程模块pip install eventlet， 执行命令celery -A celery_demo worker -l info -P eventlet

在windows10上使用celery时，异步任务的启动命令不能和定时任务的不能一块儿启动，要分开使用

celery -A celery_demo worker -l info -P eventlet

celery -A celery_demo beat -l info

联系我们 沪ICP备13005482号-10