在 Python 中,异步任务通常通过使用库如 Celery
来实现。Celery
是一个简单、灵活且可靠的分布式系统,用于处理大量消息,同时提供操作控制。
在 Celery
中,delay
和 apply_async
是两种常用的方法来调度异步任务。
delay
方法
delay
是 Celery
提供的一个快捷方法,用于简化任务的调用。它会自动将任务标记为异步执行。
from celery import Celeryapp = Celery('tasks', broker='pyamqp://guest@localhost//')@app.task
def add(x, y):return x + y# 使用 delay 方法调用任务
result = add.delay(4, 6)
apply_async
方法
apply_async
提供了更多的控制选项,例如可以指定任务的执行时间、重试策略等。
from celery import Celeryapp = Celery('tasks', broker='pyamqp://guest@localhost//')@app.task
def add(x, y):return x + y# 使用 apply_async 方法调用任务
result = add.apply_async((4, 6), countdown=10) # 任务将在10秒后执行
参数说明
args
:任务的参数,通常以元组形式传递。kwargs
:任务的关键字参数,以字典形式传递。countdown
:任务延迟执行的时间(以秒为单位)。eta
:任务的预计执行时间(datetime 对象)。expires
:任务的过期时间(datetime 对象或秒数)。retry
:是否在任务失败时自动重试。retry_policy
:重试策略,例如最大重试次数、重试间隔等。
示例1
使用 apply_async
方法来设置任务的各种参数:
from celery import Celery
from datetime import datetime, timedeltaapp = Celery('tasks', broker='pyamqp://guest@localhost//')@app.task(bind=True, max_retries=3)
def add(self, x, y):try:return x + yexcept Exception as exc:raise self.retry(exc=exc, countdown=5)# 使用 apply_async 方法调用任务
eta = datetime.utcnow() + timedelta(seconds=10)
result = add.apply_async((4, 6), eta=eta, expires=60, retry=True, retry_policy={'max_retries': 5,'interval_start': 0,'interval_step': 0.2,'interval_max': 0.2,
})
任务 add
被设置为在10秒后执行,并且在60秒后过期。如果任务失败,它会自动重试最多5次,每次重试间隔0.2秒。
delay
方法是apply_async
的简化版本,适用于简单的异步任务调用。apply_async
方法提供了更多的控制选项,适用于需要更复杂调度和重试策略的任务。
示例2
假设你有一个自定义的任务基类 CallbackTask
,你可以这样定义一个任务:
from celery import Celery, Taskapp = Celery('tasks', broker='pyamqp://guest@localhost//')class CallbackTask(Task):def on_success(self, retval, task_id, args, kwargs):print(f'Task {task_id} succeeded with result: {retval}')def on_failure(self, exc, task_id, args, kwargs, einfo):print(f'Task {task_id} failed with exception: {exc}')@app.task(name='my_custom_task', base=CallbackTask, ignore_result=True)
def add(x, y):return x + y# 调用任务
result = add.delay(4, 6)
-
自定义任务基类
CallbackTask
:on_success
方法:当任务成功完成时调用。on_failure
方法:当任务失败时调用。
-
任务定义:
@app.task(name='my_custom_task', base=CallbackTask, ignore_result=True)
:name='my_custom_task'
:任务的自定义名称。base=CallbackTask
:任务的基类是CallbackTask
。ignore_result=True
:任务的结果将不会被存储。
-
调用任务:
result = add.delay(4, 6)
:异步调用任务add
,传递参数4
和6
。
在 Celery
中,任务的参数通常以元组或字典的形式传递,并且 Celery
会自动处理参数的序列化和反序列化。因此,你通常不需要手动将参数 JSON 化。
参数传递
简单参数
from celery import Celeryapp = Celery('tasks', broker='pyamqp://guest@localhost//')@app.task
def add(x, y):return x + y# 使用 delay 方法调用任务,传递参数
result = add.delay(4, 6)
在这个示例中,参数 4
和 6
被传递给任务 add
,Celery
会自动处理这些参数的序列化和反序列化。
复杂参数
如果你需要传递更复杂的参数,例如嵌套的字典或列表,Celery
也能处理这些情况:
from celery import Celeryapp = Celery('tasks', broker='pyamqp://guest@localhost//')@app.task
def process_data(data):# 假设 data 是一个字典return data['key1'] + data['key2']# 使用 apply_async 方法调用任务,传递复杂参数
data = {'key1': 10, 'key2': 20}
result = process_data.apply_async((data,))
在这个示例中,data
是一个字典,Celery
会自动将其序列化并传递给任务 process_data
。
常见错误
在 Celery
中,如果尝试传递一个 Django 模型对象作为任务参数,而没有设置适当的序列化和反序列化方法,通常会遇到序列化错误。默认情况下,Celery
使用 JSON 作为序列化格式,而 JSON 不支持直接序列化 Django 模型对象。
如果直接传递一个 Django 模型对象作为任务参数,可能会遇到类似以下的错误:
kombu.exceptions.EncodeError: Object of type <YourModel> is not JSON serializable
这个错误表明 Celery
尝试将 Django 模型对象序列化为 JSON,但失败了,因为 JSON 序列化器不知道如何处理 Django 模型对象。
解决方法
-
传递模型对象的主键:
- 传递模型对象的主键(或其他简单类型)作为任务参数,然后在任务内部重新获取模型对象。
from celery import Celery from myapp.models import MyModelapp = Celery('tasks', broker='pyamqp://guest@localhost//')@app.task def process_model_object(model_id):obj = MyModel.objects.get(id=model_id)# 处理对象print(obj)# 调用任务,传递模型对象的主键 obj = MyModel.objects.first() process_model_object.delay(obj.id)
-
自定义序列化和反序列化:
- 自定义任务参数的序列化和反序列化方法,将模型对象转换为可序列化的格式(如字典)。
from celery import Celery from myapp.models import MyModelapp = Celery('tasks', broker='pyamqp://guest@localhost//')@app.task def process_model_object(model_data):# 反序列化模型对象obj = MyModel(**model_data)# 处理对象print(obj)# 调用任务,传递模型对象的字典表示 obj = MyModel.objects.first() model_data = {'id': obj.id,'field1': obj.field1,'field2': obj.field2,# 其他字段 } process_model_object.delay(model_data)
-
使用 Pickle 序列化器:
Celery
支持多种序列化器,包括 Pickle。Pickle 可以序列化几乎所有 Python 对象,但它有安全风险,不建议在不受信任的环境中使用。
from celery import Celery from myapp.models import MyModelapp = Celery('tasks', broker='pyamqp://guest@localhost//') app.conf.update(task_serializer='pickle',accept_content=['pickle'], # Ignore other contentresult_serializer='pickle', )@app.task def process_model_object(obj):# 处理对象print(obj)# 调用任务,传递模型对象 obj = MyModel.objects.first() process_model_object.delay(obj)
总结
- 传递模型对象的主键:这是最常见和推荐的方法,因为它简单且安全。
- 自定义序列化和反序列化:适用于需要传递复杂对象的情况。
- 使用 Pickle 序列化器:虽然方便,但有安全风险,不建议在不受信任的环境中使用。