通过前面的入门,我们大概了解了celery的工作原理及简单的入门代码示例(传送门),下面进行一些稍微复杂的任务调度学习
多目录结构异步执行
在实际项目中,使用Celery进行异步任务处理时,经常需要将代码组织在多个目录和模块中,以便更好地管理和维护。下面展示如何在多目录结构下使用Celery进行异步任务执行。
项目结构
假设我们有以下项目结构:
1. 配置文件 (config/celery_config.py
)
首先,我们需要配置Celery的broker和backend。在这个示例中,我们将配置文件放在 config
目录下。
broker_url = 'redis://localhost:32769/0'
result_backend = 'redis://localhost:32769/1'
2. 定义任务 (app/task1.py 和 app/task2.py
)
接下来,在 app
目录下定义我们的任务函数。
task1
import timefrom celery_demo.multi_task.app.celery_worker import celery_app@celery_app.task
def send_email(name):print(f"开始给{name}发送邮件")time.sleep(1)return "done"
task2
import timefrom celery_demo.multi_task.app.celery_worker import celery_app@celery_app.task
def send_msg(name):print(f"开始给{name}发送短信")time.sleep(1)return "done"
3. 启动Worker (app/celery_worker.py
)
为了能够启动worker来处理任务,我们需要编写一个脚本来启动它。在这个示例中,这个脚本放在 app/worker.py
中。
注意!:include参数的task路径必须写全,不然各种找不到模块报错
from celery import Celerycelery_app = Celery('celery_worker', include=['celery_demo.multi_task.app.task1', 'celery_demo.multi_task.app.task2'])
celery_app.config_from_object('celery_demo.multi_task.config.celery_config')# 时区
celery_app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
celery_app.conf.enable_utc = False# 启动celery,准备好接收消息,一旦接收到消息就执行任务,并存储结果
if __name__ == '__main__':celery_app.worker_main(['worker', '--loglevel=info'])
运行命令启动Worker:
在根目录下:
celery -A celery_demo.multi_task.app.celery_worker worker --loglevel=info
或者直接运行该脚本:
/usr/bin/python3 /Users/fangyirui/PycharmProjects/pythonProject/celery_demo/multi_task/app/celery_worker.py
或者pychram可以右键直接run
4. 主程序调用异步任务 (main.py
)
最后,在主程序中调用这些异步任务。我们可以通过导入 add.delay()
或者其他方法来发送消息到消息队列。
from celery_demo.multi_task.app import task1, task2result1 = task1.send_email.delay('AA')
result2 = task2.send_msg.delay('BB')print(result1)
print(result1.get())
print(result2)
print(result2.get())
总结
在上述示例中,通过合理的项目结构,将不同功能模块分离开来,使得代码更加清晰易维护。具体步骤如下:
- 配置文件:定义了Celery的broker和backend设置。
- 定义任务:创建了包含具体业务逻辑的异步函数,并用@task装饰器标记为可被调度执行的task.
- 启动Worker:编写了用于启动celerey worker 的脚本,使其能够从消息队列拉取并执行相应操作.
- 主程序调用:通过导入task 模块中的方法,实现对某些操作发起异步请求.
- 注意各种模块的导包和引用需要把路径写全,不然就算pycharm不提示报错,运行时也会报错,博主我已经踩了很多坑了!
定时任务
代码结构目录
启动beat服务方式
配置 Celery 应用
创建一个名为 celery_worker.py
的文件,并配置你的 Celery 应用。
from datetime import timedeltafrom celery import Celery
from celery.schedules import crontabapp = Celery('celery_worker', broker='redis://localhost:32769/0', backend='redis://localhost:32769/1',include=['task'])app.conf.beat_schedule = {'add-every-seconds': {# 指定执行的任务'task': 'task.add',# 单位为秒,每10秒触发一次# 'schedule': 10.0,# corn表达式触发,每一分钟触发一次# 'schedule': crontab(minute="*/1"),# 每年6月5号,16点0分执行'schedule': crontab(minute=0, hour=16, day_of_month=5, month_of_year=6),# 每6秒执行一次# 'schedule': timedelta(seconds=6),'args': (16, 16)},
}
app.conf.timezone = 'Asia/Shanghai'
定义任务
创建一个名为 task.py
的文件,并定义你的任务。
from celery_worker import app@app.task
def add(x, y):print(f"执行成功: {x} {y} ,结果为{x + y}")return x + y
启动 Worker 和 Beat 服务
启动 worker:
celery -A celery_worker worker --loglevel=info
启动 beat:
celery -A celery_worker beat --loglevel=info
这样,Celery Beat 会按照你配置的时间间隔发送 add(16, 16)
的任务到队列中,而 Worker 会从队列中取出并执行这个任务。
单个任务时指定时间
send_task.py
from datetime import datetime
from datetime import timedeltafrom task import add# # 方式一
# # 如果是过去的时间,则会立马执行
# v1 = datetime(2024, 6, 5, 16, 5, 00)
# print(v1)
# # 如果不转utc时间,会比预计的晚8个小时才执行
# v2 = datetime.utcfromtimestamp(v1.timestamp())
# print(v2)
# result = add.apply_async(args=[5, 5], eta=v2)
# print(result.id)# 方式二
ctime = datetime.now()
# 默认用utc时间
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())# 用当前时间延迟10秒执行
time_delay = timedelta(seconds=10)
task_time = utc_ctime + time_delay# 使用apply_async并设定时间
result = add.apply_async(args=[7, 7], eta=task_time)
print(result.id)
zsh: command not found: celery
问题描述
在终端执行celery的命令时报错,zsh: command not found: celery
但执行pip3 show celery
查看已经安装过的celery时,发现安装没有问题:
➜ pythonProject pip3 show celery
Name: celery
Version: 5.4.0
Summary: Distributed Task Queue.
Home-page: https://docs.celeryq.dev/
Author: Ask Solem
Author-email: auvipy@gmail.com
License: BSD-3-Clause
Location: /Users/fangyirui/Library/Python/3.9/lib/python/site-packages
Requires: billiard, python-dateutil, click-didyoumean, click, click-repl, click-plugins, vine, kombu, tzdata
Required-by:
在网上查阅其他解决方案后,问题仍没有解决,那么你很可能跟我一样还没有将 Python 二进制安装目录添加到 PATH 中。
解决方案:
在 .zshrc 文件中添加你的python/bin目录: /Users/fangyirui/Library/Python/3.9/bin
或者执行:
export PATH=$PATH:/Users/fangyirui/Library/Python/3.9/bin