Celery完成定时任务

1.什么是Celery

Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统

专注于实时处理的异步任务队列

同时也支持任务调度

celery支持linux,如果windows使用celery出了问题不解决

Celery架构

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

消息中间件

Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括，RabbitMQ, Redis等等

任务执行单元

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

任务结果存储

Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, redis等

版本支持情况

Celery version 4.0 runs on
        Python ?2.7, 3.4, 3.5?
        PyPy ?5.4, 5.5?
    This is the last version to support Python 2.7, and from the next version (Celery 5.x) Python 3.5 or newer is required.

    If you’re running an older version of Python, you need to be running an older version of Celery:

        Python 2.6: Celery series 3.1 or earlier.
        Python 2.5: Celery series 3.0 or earlier.
        Python 2.4 was Celery series 2.2 or earlier.

    Celery is a project with minimal funding, so we don’t support Microsoft Windows. Please don’t open any issues related to that platform.

2.使用场景

异步任务：将耗时操作任务提交给Celery去异步执行，比如发送短信/邮件、消息推送、音视频处理等等

定时任务：定时执行某件事情，比如每天数据统计

3.Celery的安装配置

pip install celery

消息中间件：RabbitMQ/Redis

app=Celery(‘任务名‘，backend=‘xxx‘,broker=‘xxx‘)

4.Celery执行异步任务

基本使用

创建项目celerytest

创建py文件：celery_app_task.py

import celery
import time

#消息中间件,任务提交2里面有数据
# broker=‘redis://127.0.0.1:6379/2‘ 不加密码
broker=‘redis://:[email protected]:6379/2‘
#结果存储,任务完毕1里面有数据
backend=‘redis://:[email protected]:6379/1‘

#第一个参数,给celery命名,可以是任意命名
cel=celery.Celery(‘test‘,backend=backend,broker=broker)

#假设它是一个耗时任务
#用装饰器修饰任务,这个任务才能被celery调度,否则不能调度
@cel.task
def add(x,y):
    return x+y

创建py文件：add_task.py,添加任务

from celery_app_task import add
#以前同步执行,等待结果
#result = add(4,5)
#print(result)

#现在做成异步,把任务提交到消息队列中去,用celery异步执行
#这句话只是把任务提交到消息中间件了,其实并没有执行
result = add.delay(4,5)
print(result.id)

启动工人执行任务

创建py文件：run.py，执行任务，或者使用命令执行：celery worker -A celery_app_task -l info

注：windows下：celery worker -A celery_app_task -l info -P eventlet

#windows下使用需要安装eventlet模块

from celery_app_task import cel
if __name__ == ‘__main__‘:
    cel.worker_main()
    # cel.worker_main(argv=[‘--loglevel=info‘)

创建py文件：result.py，查看任务执行结果

from celery.result import AsyncResult
from celery_app_task import cel

#id的值为add_task.py文件中打印的result.id的值
async = AsyncResult(id="e5e25734-ba93-423f-a3d4-bd87574e9be4", app=cel)

#当它值为True,表示任务执行完成,取出结果
if async.successful():
    result = async.get()
    print(result)
    # result.forget() # 将结果删除
elif async.failed():
    print(‘执行失败‘)
elif async.status == ‘PENDING‘:
    print(‘任务等待中被执行‘)
elif async.status == ‘RETRY‘:
    print(‘任务异常后正在重试‘)
elif async.status == ‘STARTED‘:
    print(‘任务已经开始被执行‘)

执行 add_task.py，添加任务，并获取任务ID

执行 run.py ，或者执行命令：celery worker -A celery_app_task -l info

执行 result.py,检查任务状态并获取结果

多任务结构

pro_cel
    ├── celery_task# celery相关文件夹
    │   ├── celery.py   # celery连接和配置相关文件,必须叫这个名字
    │   └── tasks1.py    #  所有任务函数
    │    └── tasks2.py    #  所有任务函数
    ├── check_result.py # 检查结果
    └── send_task.py    # 触发任务

celery.py

from celery import Celery

cel = Celery(‘celery_demo‘,
             broker=‘redis://:[email protected]:6379/1‘,
             backend=‘redis://:[email protected]:6379/2‘,
             # 包含以下两个任务文件，去相应的py文件中找任务，对多个任务做分类
             include=[‘celery_task.tasks1‘,
                      ‘celery_task.tasks2‘
                      ])

#如果设置了下列参数,定时任务时就不用将时间转化为UTC时间
# 时区
cel.conf.timezone = ‘Asia/Shanghai‘
# 是否使用UTC
cel.conf.enable_utc = False

tasks1.py

import time
from celery_task.celery import cel

@cel.task
def test_celery(res):
    time.sleep(5)
    return "test_celery任务结果:%s"%res

tasks2.py

import time
from celery_task.celery import cel
@cel.task
def test_celery2(res):
    time.sleep(5)
    return "test_celery2任务结果:%s"%res

check_result.py

from celery.result import AsyncResult
from celery_task.celery import cel

async = AsyncResult(id="08eb2778-24e1-44e4-a54b-56990b3519ef", app=cel)

if async.successful():
    result = async.get()
    print(result)
    # result.forget() # 将结果删除,执行完成，结果不会自动删除
    # async.revoke(terminate=True)  # 无论现在是什么时候，都要终止
    # async.revoke(terminate=False) # 如果任务还没有开始执行呢，那么就可以终止。
elif async.failed():
    print(‘执行失败‘)
elif async.status == ‘PENDING‘:
    print(‘任务等待中被执行‘)
elif async.status == ‘RETRY‘:
    print(‘任务异常后正在重试‘)
elif async.status == ‘STARTED‘:
    print(‘任务已经开始被执行‘)

send_task.py

from celery_task.tasks1 import test_celery
from celery_task.tasks2 import test_celery2

# 立即告知celery去执行test_celery任务，并传入一个参数
result = test_celery.delay(‘第一个的执行‘)
print(result.id)
result = test_celery2.delay(‘第二个的执行‘)
print(result.id)

添加任务（执行send_task.py），开启work：celery worker -A celery_task -l info -P eventlet (celery_task文件夹的名字)，检查任务执行结果（执行check_result.py）

5.Celery执行定时任务

设定时间让celery执行一个任务

add_task.py

from celery_app_task import add
from datetime import datetime

# 方式一
#设置任务执行时间,2019年4月18日18时55分56秒执行该任务
v1 = datetime(2019, 4, 18, 18, 55, 56)
print(v1)
#将时间转为utc时间
v2 = datetime.utcfromtimestamp(v1.timestamp())
print(v2)
#调用apply_async方法,args是add函数的参数,eta是时间对象
result = add.apply_async(args=[1, 3], eta=v2)
print(result.id)

# 方式二
#取出当前时间
ctime = datetime.now()
#将当前时间转化为utc时间
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
from datetime import timedelta
#延时30s
time_delay = timedelta(seconds=30)
#得到执行add函数的时间
task_time = utc_ctime + time_delay

# 使用apply_async并设定时间
result = add.apply_async(args=[4, 3], eta=task_time)
print(result.id)

类似于crontab的定时任务

作用:每年/每月/每天/每秒执行定时任务

多任务结构中celery.py修改如下

from datetime import timedelta
from celery import Celery
from celery.schedules import crontab

cel = Celery(‘tasks‘,
             broker=‘redis://:[email protected]:6379/1‘,
             backend=‘redis://:[email protected]:6379/2‘,
             # 包含以下两个任务文件，去相应的py文件中找任务，对多个任务做分类
             include=[‘celery_task.tasks1‘,
                      ‘celery_task.tasks2‘
                      ])

cel.conf.timezone = ‘Asia/Shanghai‘
cel.conf.enable_utc = False

cel.conf.beat_schedule = {
    # 名字随意命名
    ‘add-every-2-seconds‘: {
        # 执行tasks1下的test_celery函数
        ‘task‘: ‘celery_task.tasks1.test_celery‘,

        # 每隔2秒执行一次
        ‘schedule‘: timedelta(seconds=2),

        # 传递参数
        ‘args‘: (‘给函数要传递的参数‘)
    },

    #名字随意命名
    ‘add-every-12-seconds‘: {
         ‘task‘: ‘celery_task.tasks1.test_celery‘,
    #     每年4月11号，8点42分执行
         ‘schedule‘: crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
         ‘args‘: (‘给函数要传递的参数‘)
     },
}

#在pro_cel文件夹下执行
#发送任务
启动一个beat：celery beat -A celery_task -l info

#执行任务
启动work执行：celery worker -A celery_task -l info -P  eventlet

#在cmd中退出任务 ctrl+c

6.Django中使用Celery

在Python脚本中调用Django环境(Celery是一个脚本)

import os
if __name__ == ‘__main__‘:
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "untitled15.settings")
    import django
    django.setup()

    from app01 import models

    books = models.Book.objects.all()
    print(books)

安装包

版本特别重要,对上才能用,对不上各种出错
celery==3.1.25
django-celery==3.1.20

在项目目录下创建celeryconfig.py

import djcelery
djcelery.setup_loader()

#其中任务可以写多个
CELERY_IMPORTS=(
    ‘app01.tasks‘,
)
#有些情况可以防止死锁
CELERYD_FORCE_EXECV=True
# 设置并发worker数量
CELERYD_CONCURRENCY=4
#允许重试
CELERY_ACKS_LATE=True
# 每个worker最多执行100个任务被销毁，可以防止内存泄漏
CELERYD_MAX_TASKS_PER_CHILD=100
# 超时时间
CELERYD_TASK_TIME_LIMIT=12*30

在app01目录下创建tasks.py

from celery import task
@task
def add(a,b):
    return a+b

视图函数views.py

from django.shortcuts import render,HttpResponse
from app01.tasks import add
from datetime import datetime
def test(request):
    ctime = datetime.now()
    # 默认用utc时间
    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
    from datetime import timedelta
    time_delay = timedelta(seconds=5)
    task_time = utc_ctime + time_delay
    result = add.apply_async(args=[4, 3], eta=task_time)
    print(result.id)
    return HttpResponse(‘ok‘)

路由层url.py

from app01 import views
urlpatterns = [
    url(r‘^test/$‘, views.test),
]

settings.py

INSTALLED_APPS = [
    ...
    ‘djcelery‘,
    ‘app01‘
]

...

from djagocele import celeryconfig
BROKER_BACKEND=‘redis‘
#BOOKER_URL=‘redis://127.0.0.1:6379/1‘ 不加密
BOOKER_URL=‘redis://:[email protected]:6379/1‘
#CELERY_RESULT_BACKEND=‘redis://127.0.0.1:6379/2‘ 不加密
CELERY_RESULT_BACKEND=‘redis://:[email protected]:6379/2‘

原文地址：https://www.cnblogs.com/lizeqian1994/p/10751514.html

时间： 2024-08-02 22:27:13

Celery完成定时任务的相关文章

django+celery配置(定时任务)

下面介绍一下django+celery的配置做定时任务 1.首先介绍一下环境和版本 python==2.7 django == 1.8.1 celery == 3.1.23 django-celery == 3.1.17 2.celery的安装   sudo pip install celery==3.1.23 sudo pip install django-celery==3.1.17 3.新建一个项目 (1)django-admin startproject django_celery_de

Django中使用Celery实现定时任务(用djcelery)

[TOC] 一.引言 Django是python语言下的一个比较热门的Web框架,越来越多的企业和开发者使用Django实现自己的Web服务器.在Web服务器开发过程中,有时候我们不仅仅是要实现Web服务器端和用户端的简单逻辑交互,还要实现一些定时任务.举出以下的例子: 定期删除或缓存Redis数据库的记录 为了追求更高的数据库访问性能,我把Redis作为MySql数据库的缓存.把常访问的数据放在Redis中,然后定时存储到Mysql中.并且把过期的Redis数据删掉．那么这个时候,就需要定时去

Celery Beat定时任务

示例代码 创建schedule_tasks.py文件 from celery import Celery from celery.schedules import crontab app = Celery("SchedulerTasks", broker='pyamqp://admin:[email protected]:5672//') @app.on_after_configure.connect def setup_periodic_tasks(sender, **kwargs)

44. Python Celery多实例 定时任务

celery是一个分布式的任务调度模块,那么celery是如何和分布式挂钩呢? celery可以支持多台不同的计算机执行不同的任务或者相同的任务. 如果要说celery的分布式应用的话,就要提到celery的消息路由机制,提到AMQP协议. 具体可以查看AMQP文档详细了解. 简单理解: 可以有多个"消息队列"(message Queue),不同的消息可以指定发送给不同的Message Queue, 而这是通过Exchange来实现的,发送消息到"消息队列"中时,可

Celery学习--- Celery操作之定时任务

celery支持定时任务,设定好任务的执行时间,celery就会定时自动帮你执行, 这个定时任务模块叫celery beat 文件定时执行任务 项目前提: 安装并启动Redis celery_Schedule.py # version: python3.2.5 # author: 'lft415659' from celery import Celery from celery.schedules import crontab # 可以实现复杂的定时任务 app = Celery('Celery

python celery多worker、多队列、定时任务

多worker.多队列 celery是一个分布式的任务调度模块,那么怎么实现它的分布式功能呢,celery可以支持多台不同的计算机执行不同的任务或者相同的任务. 如果要说celery的分布式应用的话,就要提到celery的消息路由机制,提到AMQP协议. 简单理解: 可以有多个"消息队列"(message Queue),不同的消息可以指定发送给不同的Message Queue, 而这是通过Exchange来实现的,发送消息到"消息队列"中时,可以指定routiing

celery执行异步任务和定时任务

一.什么是Clelery Celery是一个简单.灵活且可靠的,处理大量消息的分布式系统 专注于实时处理的异步任务队列 同时也支持任务调度 Celery架构 Celery的架构由三部分组成,消息中间件(message broker),任务执行单元(worker)和任务执行结果存储(task result store)组成. 消息中间件 Celery本身不提供消息服务,但是可以方便的和第三方提供的消息中间件集成.包括,RabbitMQ, Redis等等 任务执行单元 Worker是Celery提供

celery 定时任务

1 基本概念 使用 Celery 实现定时任务的步骤: (1) 创建一个 Celery 实例 (2) 配置文件中配置任务 ,发布任务 celery A xxx beat (3) 启动 Celery Worker (4) 存储结果 使用 Celery 实现异步任务的步骤: (1) 创建一个 Celery 实例 (2) 启动 Celery Worker ,通过delay() 或 apply_async()(delay 方法封装了 apply_async, apply_async支持更多的参数 ) 将

celery 定时任务，使用crontab表达式不执行(版本4.3.x）

celery 定时任务,使用crontab表达式不执行(版本4.3.x) 在使用celery 执行定时任务时,发现任务不会执行,schedule设置如下: 经测试,如果去掉hour,则任务每分钟都会执行,说明是hour的问题,那只有是时区的问题了. 遂将hour改为UTC的时间,发现任务可以执行,说明celery使用的是UTC时区,但我设置的时区如下: CELERY_TIMEZONE='Asia/Shanghai' CELERY_ENABLE_UTC=True 注意添加任务的时候使用utc时间,