首页 > 其他分享 > 【celery详解】celery学习md笔记 第（2）期：Celery任务调度详解

【celery详解】celery学习md笔记 第（2）期：Celery任务调度详解

时间：2023-11-13 21:32:13浏览次数：43  
标签：task app django celery 任务 详解 任务调度 my

Celery是一个功能完备即插即用的任务队列。它使得我们不需要考虑复杂的问题，使用非常简单。celery看起来似乎很庞大，本文我们先对其进行简单的了解，然后再去学习其他一些高级特性。

全套笔记直接地址： 请移步这里

共 4 章，12 子模块

介绍一下如何调用任务，队列路由.

1. signature

我们到目前为止只是学习了如何使用delay()方法，当然这个方法也是非常常用的。但是有时我们并不想简单的将任务发送到队列中，我们想将一个任务函数(由参数和执行选项组成)作为一个参数传递给另外一个函数中，为了实现此目标，Celery使用一种叫做signatures的东西。

一个signature包装了一个参数和执行选项的单个任务调用。我们可将这个signature传递给函数。

我们先看下tasks.py模块中定义的任务函数:

from proj.celery import app as celery_app

  
  
# 创建任务函数
  
  
@celery_app.task
def my_task1():
    print("任务函数(my_task1)正在执行....")

@celery_app.task
def my_task2():
    print("任务函数(my_task2)正在执行....")

@celery_app.task
def my_task3():
    print("任务函数(my_task3)正在执行....")

我们将my_task1()任务包装称一个signature:

t1 = my_task1.signatures(countdown=10)
t1.delay()

2. Primitives

这些primitives本身就是signature对象，因此它们可以以多种方式组合成复杂的工作流程。primitives如下:

group: 一组任务并行执行，返回一组返回值，并可以按顺序检索返回值。

chain: 任务一个一个执行，一个执行完将执行return结果传递给下一个任务函数.

tasks.py模块如下:

from proj.celery import app as celery_app

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

@celery_app.task
def my_task2(a, b):
    print("任务函数(my_task2)正在执行....")
    return a + b

@celery_app.task
def my_task3(a, b):
    print("任务函数(my_task3)正在执行....")
    return a + b

group案例如下(test.py模块):

from proj.tasks import my_task1
from proj.tasks import my_task2
from proj.tasks import my_task3
from celery import group

  
  
# 将多个signature放入同一组中
  
  
my_group = group((my_task1.s(10, 10), my_task2.s(20, 20), my_task3.s(30, 30)))
ret = my_group() # 执行组任务
print(ret.get())  # 输出每个任务结果

chain案例如下(test.py模块):

from proj.tasks import my_task1
from proj.tasks import my_task2
from proj.tasks import my_task3
from celery import chain

  
  
# 将多个signature组成一个任务链
  
  
  
  
# my_task1的运行结果将会传递给my_task2
  
  
  
  
# my_task2的运行结果会传递给my_task3
  
  
my_chain = chain(my_task1.s(10, 10) | my_task2.s(20) | my_task3.s(30))
ret = my_chain()  # 执行任务链
print(ret.get())  # 输出最终结果

假如我们有两个worker,一个worker专门用来处理邮件发送任务和图像处理任务，一个worker专门用来处理文件上传任务。

我们创建两个队列，一个专门用于存储邮件任务队列和图像处理，一个用来存储文件上传任务队列。

Celery支持AMQP(Advanced Message Queue)所有的路由功能，我们也可以使用简单的路由设置将指定的任务发送到指定的队列中.

我们需要配置在celeryconfig.py模块中配置 CELERY_ROUTES 项, tasks.py模块修改如下:

from proj.celery import app as celery_app


@celery_app.task
def my_task1(a, b):
    print("my_task1任务正在执行....")
    return a + b


@celery_app.task
def my_task2(a, b):
    print("my_task2任务正在执行....")
    return a + b


@celery_app.task
def my_task3(a, b):
    print("my_task3任务正在执行....")
    return a + b


@celery_app.task
def my_task4(a, b):
    print("my_task3任务正在执行....")
    return a + b


@celery_app.task
def my_task5():
    print("my_task5任务正在执行....")


@celery_app.task
def my_task6():
    print("my_task6任务正在执行....")


@celery_app.task
def my_task7():
    print("my_task7任务正在执行....")

我们通过配置，将send_email和upload_file任务发送到queue1队列中，将image_process发送到queue2队列中。

我们修改celeryconfig.py:

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


task_routes=({
    'proj.tasks.my_task5': {'queue': 'queue1'},
    'proj.tasks.my_task6': {'queue': 'queue1'},
    'proj.tasks.my_task7': {'queue': 'queue2'},
    },
)

test.py:

from proj.tasks import *

  
  
# 发送任务到路由指定的队列中

my_task5.delay() my_task6.delay() my_task7.delay()

# 开启两个worker服务器，分别处理两个队列：

celery -A proj worker --loglevel=info -Q queue1
celery -A proj worker --loglevel=info -Q queue2

我们同样也可以通过apply_aynsc()方法来设置任务发送到那个队列中:

my_task1.apply_async(queue='queue1')

我们也可设置一个worker服务器处理两个队列中的任务:

celery -A proj worker --loglevel=info -Q queue1,queue2

celery beat是一个调度器，它可以周期内指定某个worker来执行某个任务。如果我们想周期执行某个任务需要增加beat_schedule配置信息.

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

  
  
# 指定任务发到那个队列中
  
  
task_routes=({
    'proj.tasks.my_task5': {'queue': 'queue1'},
    'proj.tasks.my_task6': {'queue': 'queue1'},
    'proj.tasks.my_task7': {'queue': 'queue2'},
    },
)


  
  
# 配置周期性任务，　或者定时任务
  
  
beat_schedule = {
    'every-5-seconds':
        {
            'task': 'proj.tasks.my_task8',
            'schedule': 5.0,
            # 'args': (16, 16),
        }
}

tasks.py模块内容如下:

from proj.celery import app as celery_app


@celery_app.task
def my_task1(a, b):
    print("my_task1任务正在执行....")
    return a + b


@celery_app.task
def my_task2(a, b):
    print("my_task2任务正在执行....")
    return a + b


@celery_app.task
def my_task3(a, b):
    print("my_task3任务正在执行....")
    return a + b


@celery_app.task
def my_task4(a, b):
    print("my_task3任务正在执行....")
    return a + b


@celery_app.task
def my_task5():
    print("my_task5任务正在执行....")




@celery_app.task
def my_task6():
    print("my_task6任务正在执行....")



@celery_app.task
def my_task7():
    print("my_task7任务正在执行....")


  
  
# 周期执行任务
  
  
@celery_app.task
def my_task8():
    print("my_task8任务正在执行....")

启动woker处理周期性任务:

celery -A proj worker --loglevel=info --beat

如果我们想指定在某天某时某分某秒执行某个任务，可以执行cron任务, 增加配置信息如下:

beat_schedule = {
    'every-5-minute':
        {
            'task': 'proj.tasks.period_task',
            'schedule': 5.0,
            'args': (16, 16),
        },
    'add-every-monday-morning': {
        'task': 'proj.tasks.period_task',
        'schedule': crontab(hour=7, minute=30, day_of_week=1),
        'args': (16, 16),
    },

}

crontab例子: http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html

开启一个celery beat服务：

celery -A proj beat

celery需要保存上次任务运行的时间在数据文件中，文件在当前目录下名字叫celerybeat-schedule. beat需要访问此文件：

celery -A proj beat -s /home/celery/var/run/celerybeat-schedule

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://:[email protected]: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("服务器返回响应内容!")

在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://:[email protected]: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://:[email protected]: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秒中就会执行一次，如果配置了存储，那么每次任务执行的结果也会被保存到对应的数据库中。

未完待续 下一期下一章

全套笔记直接地址： 请移步这里

标签：task,app,django,celery,任务,详解,任务调度,my
From： https://blog.51cto.com/u_13578013/8353346

相关文章

  • Angular 服务器端渲染应用 re-hydration 过程详解
    当使用Angular启用服务器端渲染（Server-SideRendering，以下简称SSR）时，应用程序的工作方式发生了显著变化。这使得Angular应用更加友好，不仅对搜索引擎爬虫更友好，还有助于改善应用的性能和加载时间。在本文中，我们将详细介绍在浏览器端访问启用SSR的Angular应用时背后发生的事情，特别侧......
  • timestamp(6)详解 在MySQL中，timestamp是一种时间戳类型。timestamp(6)是timestamp类型
    timestamp(6)详解在MySQL中，timestamp是一种时间戳类型。timestamp(6)是timestamp类型的一个子类型，表示精确到秒后6位小数的时间戳。它占用8个字节存储空间一、什么是timestamp(6)在MySQL中，timestamp是一种时间戳类型。timestamp(6)是timestamp类型的一个子类型，表示精确到秒后6......
  • .NET Web API 详解
    当涉及到.NETWebAPI时，这是Microsoft提供的一个框架，用于构建可以被各种客户端消费的HTTP服务，包括浏览器、移动设备和桌面应用程序。它是ASP.NET平台的一部分，旨在简化构建RESTfulWebAPI的过程。以下是.NETWebAPI的一些关键概念和特性：RESTful服务：WebAPI建立在Representational......
  • celery
    介绍#celery是什么？ 分布式异步任务框架：第三方框架，celery翻译过来是芹菜，吉祥物就是芹菜项目中使用异步任务的场景，可以使用它之前做异步，如何做？异步发送短信---》开启多线程---》不便于管理#celery有什么作用？ -执行异步任务-执行延迟任务-执行定时任务......
  • kubectl详解
    kubectl详解陈述式资源管理方法：1.kubernetes集群管理集群资源的唯一入口是通过相应的方法调用apiserver的接口2.kubectl是官方的CLI命令行工具，用于与apiserver进行通信，将用户在命令行输入的命令，组织并转化为apiserver能识别的信息，进而实现管理k8s各种资源的一种有效......
  • sar命令详解（系统活动报告）
    首页新闻博问会员闪存班级    HZhuizai 博客园首页新随笔联系订阅管理 linux分析利刃之sar命令详解 目录 一、sar的概述二、统计CPU使用情况三、平均负载统计分析四、内存统计分析五、统计swap......
  • Java核心知识体系6：集合框架详解
    Java核心知识体系1：泛型机制详解Java核心知识体系2：注解机制详解Java核心知识体系3：异常机制详解Java核心知识体系4：AOP原理和切面应用Java核心知识体系5：反射机制详解1集合框架图总览我们来简单解读下上面这个框架图：所有集合类都位于java.util包下Iterator是遍历集合的工具......
  • jUnit测试框架入门详解​(加强版)
    jUnit测试框架入门详解入门知识什么是单元测试单元测试是针对最小的功能单元编写的测试代码。Java程序最小的功能单元是方法，因此单元测试就是针对单个Java方法的测试。为什么要使用单元测试使用main()方法测试的缺点：只能有一个main()方法,不能把测试代码分离，且也没有打印出测试结果......
  • 神经网络入门篇：详解多样本向量化（Vectorizing across multiple examples）
    多样本向量化与上篇博客相联系的来理解逻辑回归是将各个训练样本组合成矩阵，对矩阵的各列进行计算。神经网络是通过对逻辑回归中的等式简单的变形，让神经网络计算出输出值。这种计算是所有的训练样本同时进行的，以下是实现它具体的步骤：图1.4.1上篇博客中得到的四个等式。它们......
  • （十三）C#编程基础复习——函数/方法详解
    C#中的函数（也可以称为方法）是一段具有签名（由函数名、参数类型和参数修饰符组成的函数信息）的代码块，可以用来实现特定的功能。一般情况下一个函数由以下几个部分组成：访问权限修饰符：用于指定函数对一个类的可见性；（public  公开的，公共的，同一程序集的其他任何代码或引用该程序集的其......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99