【Python】Celery简单入门

1. 下载celery

pip install celery

1. 下载redis

github网址下载

1. 基本写法

导入包

# 导入celery
from celery import Celery
# 导入计时器
from celery.schedules import crontab
import datetime

实例化Celery

# 固定写法
app = Celery('tasks')

加载配置项

# 固定写法
app.config_from_object(Config)

写配置

class Config():
    # 使用redis 7号数据库存储任务队列
    broker_url = 'redis://127.0.0.1:6379/7'
    # 使用redis 8号数据库存储任务结果
    result_backend = 'redis://127.0.0.1:6379/8'
    # 设置时区 没有北京时区
    timezone = 'Asia/Shanghai'
    # 配置任务队列
    beat_schedule = {
        '任务名称':{
            'task':'文件名.func'，
            'schedule':crontab() # 默认每分钟执行
            # 这里除了文件名 函数名 crontab基本都是写死的
        }
    }

crontab函数的参数：

写装饰器函数(自动化任务)

@app.task
def func():
    print('执行任务')
    print(datetime.datetime.now())

在终端发布任务 ：celery -A 文件名 beat

在终端执行任务:celery -A 文件名 worker

如果要执行的文件在某文件夹里，记得cd进去

完整代码如下：

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


class Config():
    # 使用redis 7号数据库存储任务队列
    broker_url = 'redis://127.0.0.1:6379/7'
    # 使用redis 8号数据库存储执行结果
    result_backend = 'redis://127.0.0.1:6379/8'
    # 设置时区 没有北京
    timezone = 'Asia/Shanghai'

    # 任务队列配置
    beat_schedule = {
        '任务名称':{
            'task':'celery-test.func',
            'schedule':crontab(minute='*/1') 
        }
    }



# 实例化
app = Celery('tasks')
# 加载配置项
app.config_from_object(Config)

# 自动化的任务
@app.task
def func():
    print('执行任务')
    print(datetime.datetime.now())