python之异步任务

在 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)

1. 自定义任务基类 CallbackTask：

   • on_success 方法：当任务成功完成时调用。
   • on_failure 方法：当任务失败时调用。

2. 任务定义：

   • @app.task(name='my_custom_task', base=CallbackTask, ignore_result=True)：
     • name='my_custom_task'：任务的自定义名称。
     • base=CallbackTask：任务的基类是 CallbackTask。
     • ignore_result=True：任务的结果将不会被存储。

3. 调用任务：

   • 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 模型对象。

解决方法

1. 传递模型对象的主键：

   • 传递模型对象的主键（或其他简单类型）作为任务参数，然后在任务内部重新获取模型对象。

   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)
   

2. 自定义序列化和反序列化：

   • 自定义任务参数的序列化和反序列化方法，将模型对象转换为可序列化的格式（如字典）。

   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)
   

3. 使用 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 序列化器：虽然方便，但有安全风险，不建议在不受信任的环境中使用。