Python Celery实战:异步、定时与周期任务完整指南

Chloe
Chloe
发布时间:2026-10-13 | 2人浏览过
原创

本文详解Python中Celery库的使用,涵盖异步、定时及周期任务的配置与实现。通过解析Broker、Worker与Backend架构,结合Linux与Windows环境下的启动命令,展示如何基于Redis实现任务调度。内容包含delay异步调用、apply_async定时执行及crontab周期任务的具体代码示例,帮助开发者快速掌握分布式任务队列的核心用法。

Celery异步任务、定时与周期任务实战

本文主要介绍如何在Python中利用Celery实现异步任务、周期任务及定时任务。文中通过详细的示例代码进行说明,内容详实,对读者的学习或工作具有实际的参考价值,建议相关读者仔细阅读并参考。

大家好,好久不见,这次带来的是Celery在Python中的实际应用,以及设置异步任务、周期任务和定时任务的具体步骤,希望能给刚接触这一领域的你提供一些帮助。

Celery核心架构与工作流程

首先介绍Celery的基本概念。Celery是一个专注于实时处理和调度任务的分布式任务队列,它提供了操作和维护分布式系统所需的全部数据支持。因此,你可以利用其提供的接口快速实现并管理分布式任务队列。需要注意的是,Celery本身并非任务队列,而是封装了对常见任务队列的各种操作,允许用户快速进行任务队列的使用与管理。在Python环境中,Celery主要由三部分组成:1. 用户任务应用(app);2. 管道(broker),用于存储任务,官方推荐使用Redis或RabbitMQ;3. 后端(backend),用于存储任务执行结果;4. 员工(worker),负责执行任务。大致流程如下:

在最左侧是用户,用户向服务器发起一个请求,要求服务器执行10个任务。这10个任务被分发给10个调度器,即开启10个线程进行任务处理。Worker会持续监听调度器是否有新任务,一旦发现新任务,便会立即执行。任务执行完毕后,结果会返回给调度器(即backend),Backend随后将请求发送给服务器,服务器再将结果返回给用户。最终表现是这10个任务同时完成并返回。这就是Celery的整个工作流程,其中的角色分别为:任务(app_work)、调度器(broker + backend,负责缓存任务,相当于生产者)、消费者(Worker,可指定数量,即在创建Worker命令时指定数量)。在Worker拿到任务后,用户无法再控制其执行过程,除非杀死Worker进程,否则任务必定会执行完毕。

简而言之,流程为:任务到达后,调度器(Broker)负责缓存任务,Worker负责执行任务,执行完成后将结果返回给Backend,接着返回给用户。

此外,关于在Linux上为何不使用系统自带的定时任务来处理周期和定时任务,主要是因为Linux自带的周期定时任务不可控且难以管理,保存返回值也较为麻烦。而使用Celery时,只要调度器保持开启状态,就可以随时获取任务结果,因此使用Celery进行控制非常便捷。

异步任务实现与代码示例

接下来是具体的实例代码展示:

workers.py文件内容如下:


from celery import Celery
import time
# 创建一个Celery实例, 就是用户的应用app 第一个参数是任务名称, 可以随意起 后面的就是配置的broker和backend
diaoduqi= Celery("mytask", broker="redis://127.0.0.1:6379", backend="redis:127.0.0.1:6379")
# 接下来是为应用创建任务 ab
@diaoduqi.task
def ab(a,b):
  time.sleep(15)
  return a+b

brokers.py文件内容如下:


from worker import ab

# 将任务交给Celery的Worker执行
res = ab.delay(2,4)

#返回任务ID
print(res.id)

backends.py文件内容如下:


from celery.result import AsyncResult
from worker import diaoduqi

# 异步获取任务返回值
async_task = AsyncResult(id="31ec65e8-3995-4ee1-b3a8-1528400afd5a",app=diaoduqi)

# 判断异步任务是否执行成功
if async_task.successful():
  #获取异步任务的返回值
  result = async_task.get()
  print(result)
else:
  print("任务还未执行完成")

为了简化说明,现在直接将三个文件代表的部分命名在文件名中。首先是启动workers.py。

启动方式依据系统不同而有所区别。在Linux下,执行命令 `celery worker -A workers -l INFO` 即可启动。也可以指定开启的Worker数量,即在后面添加参数 `-c 5` 表示指定5个Worker。理论上指定的Worker数量无上限。

在Windows环境下,需要安装 `eventlet` 模块才能运行成功,否则无法执行。可以通过 `pip install eventlet` 安装以开启线程支持。不指定数量时默认开启6个Worker,理论上Worker数量可以开启无限个。具体命令为 `celery worker -A s1 -l INFO -P eventlet -c 5`,使用eventlet开启5个Worker执行。

执行该命令后,系统处于就绪状态,此时需要发布任务,即通过brokers.py进行任务发布。方法是使用 `delay` 方式执行异步任务,该方法返回一个任务ID。接着在backends.py中获取这个任务ID,去查询任务是否完成。判定条件即通过 `task.successful` 判断是否执行完毕。上述内容即为Celery异步执行任务的用法与解释。

项目中的Celery应用与目录结构

接下来介绍Celery在项目中的实际应用。

在实际项目中应用Celery遵循一定的规则,目录结构应如下所示。

结构说明:首先创建一个名为CeleryTask的包,接着在其中创建一个 `celery.py` 文件,该文件名必须固定。关于重名问题,模块查找顺序是从当前目录中寻找,若根本找不到,再从内置模块中查找。这里编写的celery文件即属于此类情况。

celery.py文件内容如下:


from celery import Celery
DDQ = Celery("DDQ",broker="redis://127.0.0.1:6379",backend="redis://127.0.0.1:6379",
       include=["CeleryTask.TaskOne","CeleryTask.TaskTwo"])

TaskOne.py文件内容如下:


import time
from CeleryTask.celery import DDQ
@DDQ.task
def one1(a,b):
  # time.sleep(3)
  return a+b
@DDQ.task
def one2():
  time.sleep(2)
  return "one2"

taskTwo.py文件内容如下:


import time
from CeleryTask.celery import DDQ
@DDQ.task
def two1():
  time.sleep(2)
  return "two1"
@DDQ.task
def two2():
  time.sleep(3)
  return "two2"

getR.py文件内容如下:


from CeleryTask.TaskOne import one1 as one

# one.delay(10,10)
# two.delay(20,20)

# 定时任务我们不在使用delay这个方法了,delay是立即交给task 去执行
# 现在我们使用apply_async定时执行

# 首先我们要先给task一个执行任务的时间
import datetime, time

# 获取当前时间 此时间为东八区时间
ctime = time.time()
# 将当前的东八区时间改为 UTC时间 注意这里一定是UTC时间,没有其他说法
utc_time = datetime.datetime.utcfromtimestamp(ctime)
# 为当前时间增加 10 秒
add_time = datetime.timedelta(seconds=10)
action_time = utc_time + add_time

# action_time 就是当前时间未来10秒之后的时间
# 现在我们使用apply_async定时执行
res = one.apply_async(args=(6, 10), eta=action_time)
res = one.apply_async(args=(6, 10), eta=action_time)
res = one.apply_async(args=(6, 10), eta=action_time)
res = one.apply_async(args=(6, 10), eta=action_time)
res = one.apply_async(args=(6, 10), eta=action_time)
res = one.apply_async(args=(6, 10), eta=action_time)
print(res.id)
# 这样原本延迟5秒执行的One函数现在就要在10秒钟以后执行了

定时任务配置与执行

接着在命令行中cd到与CeleryTask同级目录下,使用命令 `celery worker -A CeleryTask -l INFO -P eventlet -c 50` 这样便开启了Worker。接着去发布任务,在定时任务中不再使用 `delay` 这个方法了。

`delay` 是立即交给Task去执行,在这里使用 `apply_async` 进行定时执行,指的是在调度时去定时执行。

需要设置的是UTC时间,以及定时的时间(多长时间以后执行)。之后使用 `celery worker -A CeleryTask -l INFO -P eventlet -c 50` 命令开启Worker,之后运行getR.py文件发布任务,可以看到在定义的时间以后执行该任务。

周期任务配置与Beat调度

周期任务部分。

周期任务指的是在指定时间去执行任务,需要导入的一个模块有 `crontab`。

文件结构如下:

结构同定时任务差不多,只不过需要变动一下文件内容。GetR文件已经不需要了,可以删除。

celery.py文件内容如下:


from celery import Celery
from celery.schedules import crontab

DDQ = Celery("DDQ", broker="redis://127.0.0.1:6379", backend="redis://127.0.0.1:6379",
       include=["CeleryTask.TaskOne", "CeleryTask.TaskTwo"])

# 我要要对beat任务生产做一个配置,这个配置的意思就是每10秒执行一次Celery_task.task_one任务参数是(10,10)
DDQ.conf.beat_schedule = {
  "each10s_task": {
    "task": "CeleryTask.TaskOne.one1",
    "schedule": 10, # 每10秒钟执行一次
    "args": (10, 10)
  },
  "each1m_task": {
    "task": "CeleryTask.TaskOne.one2",
    "schedule": crontab(minute=1) # 每1分钟执行一次 也可以替换成 60 即 "schedule": 60
  }
}

TaskOne.py文件内容如下:


import time
from CeleryTask.celery import DDQ
@DDQ.task
def one1(a,b):
  # time.sleep(3)
  return a+b
@DDQ.task
def one2():
  time.sleep(2)
  return "one2"

taskTwo.py文件内容如下:


import time
from CeleryTask.celery import DDQ
@DDQ.task
def two1():
  time.sleep(2)
  return "two1"
@DDQ.task
def two2():
  time.sleep(3)
  return "two2"

以上配置完成以后,这时候就不能直接创建Worker了,因为要执行周期任务,需要首先有一个任务的生产方,即执行 `celery beat -A CeleryTask`,用来产生调度者(Beat)。接着是创建Worker,Worker的创建命令还是原来的命令,即 `celery worker -A CeleryTask -l INFO -P eventlet -c 50`。创建完Worker之后,每10秒就会由Beat创建一个任务给Worker去执行。至此,Celery创建异步任务、周期任务、定时任务完毕,伙伴们自己拿去测试吧。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持本站。

本站声明

本文内容仅供参考,不代表本站观点。转载或引用请注明出处。如发现内容问题,请通过联系我们页面反馈。

Python 相关产品和应用

全部下载

专题列表