在第一篇文章中，我们介绍了Celery的基本需求，delay()方法的工作原理，以及通过实际使用示例来提升您对异步任务的理解。delay()是一个强大的工具，可以便捷地将任务丢入队列，但有时您可能需要更细致的控制。正是在这种情况下，今天我们将深入探讨的apply_async()方法便脱颖而出。

今天，我们将全面了解apply_async()是什么，delay()与其有什么不同，以及如何利用各种选项更有效地管理异步任务。

Celery 像卡通形象一样努力工作

1. apply_async()，打开更多可能性的工具

apply_async()是另一个强大的方法，用于将Celery任务调度为异步执行。乍看之下，它与delay()相似，但提供了更多的选项，例如任务执行时间、方式及队列设置等，为开发者带来了更大的灵活性。

请记住，delay()只是以apply_async(args=args)的形式调用的包装函数，因此可以把apply_async()看作是包含了delay()所有功能的扩展版，提供了更多的控制权。

2. apply_async()的工作原理：精细调整

apply_async()的基本工作方式与delay()相似。其核心流程是将包含任务名称和参数的消息发送给消息代理，然后由工作进程接收并执行。然而，apply_async()允许通过多种参数(parameters)精细调控任务的执行方式。

主要参数如下：

  • args：传递给任务函数的位置参数（positional arguments）的列表或元组。
  • kwargs：传递给任务函数的关键字参数（keyword arguments）的字典。
  • countdown：指定任务在延迟指定秒数后执行，而不是立即执行。例如，设置countdown=60，任务将在调用时延迟60秒后被添加到队列。
  • eta（预计到达时间）：将任务调度在特定的未来时刻执行。以datetime对象形式指定。countdown和eta是互斥的。
  • expires：设置任务的过期时间。一旦超过指定时间，工作进程将忽略该任务而不执行。可指定为int（秒）、float（秒）或datetime对象，适合临时数据或有截止日期的任务。
  • queue：将任务路由到特定名称的队列。Celery可以设置多个队列，根据任务的类型或重要性在不同的工作组中处理。
  • routing_key：根据消息代理的路由规则将任务发送至特定队列。在使用RabbitMQ等消息代理的高级路由设置时使用。
  • priority：设置任务的优先级。根据代理和工作设置，优先级较高的任务可能会优先处理。通常有整数值从0（最高优先级）到9（最低优先级）。
  • serializer：指定用于序列化任务消息的方法（如'json'、'pickle'、'yaml'等）。
  • compression：指定用于压缩任务消息的方法（如'gzip'、'bzip2'等）。
  • headers：可以在任务消息中包含附加的头信息。
  • link：指定当前任务成功完成后将执行的回调任务(callback task)。用于实现链式调用（chaining）。
  • link_error：指定当前任务失败后将执行的错误回调任务(error callback task)。

通过如此多的参数，apply_async()能够比delay()更加精细地控制任务的执行时机，路由任务到特定工作组，并管理任务的优先级，以满足复杂的异步处理需求。

3. apply_async()使用示例：情境应用

现在，让我们通过不同选项的使用示例，亲自验证apply_async()的强大。

示例 1：在特定时间后执行（countdown）

假设用户注册后需要在3天后发送欢迎邮件：

send_welcome_email.apply_async(args=[user.email], countdown=3 * 24 * 60 * 60)

使用countdown选项时，任务不会立即添加到队列，而是在指定时间（3天）后添加并由工作进程处理。

示例 2：在特定时刻执行（eta）

如果需要在每天凌晨3点执行日常报告生成任务：

from datetime import datetime, timedelta

target_time = datetime.now().replace(hour=3, minute=0, second=0, microsecond=0) + timedelta(days=1) # 次日凌晨3点

generate_daily_report.apply_async(eta=target_time)

通过eta选项，可以准确预定任务在指定时间执行。

示例 3：设置过期时间（expires）

如果外部API调用结果仅在10分钟内有效，之后不再需要执行该任务：

from datetime import datetime, timedelta

expires_at = datetime.now() + timedelta(minutes=10)

fetch_external_data.apply_async(args=[api_url], expires=expires_at)

通过expires选项，可以避免任务不必要的执行，从而节省资源。

示例 4：路由到特定队列（queue）

当想要将CPU密集型的图像处理任务在单独的工作组中处理时：

process_image.apply_async(args=[image_id], queue='image_processing')

通过Celery设置，可以单独配置监控image_processing队列的工作进程，从而确保图像处理任务仅在该工作组中处理。

示例 5：连接回调任务（link）

假设希望在支付处理任务成功完成后自动执行发送通知邮件的任务：如果不使用apply_async，可能会像下面这样在函数内部直接调用.delay()来发送邮件：

def process_payment(order_id):
    # 编写支付处理逻辑
    send_notification_email.delay(order_id)

但如果使用apply_async()，可以这样编写：

@shared_task
def process_payment(order_id):
    # 支付处理逻辑
    return order_id

@shared_task
def send_notification_email(order_id):
    # 发送通知邮件逻辑

payment_result = process_payment.apply_async(args=[order_id], link=send_notification_email.s())

此处，.s()用于生成send_notification_email任务的签名(signature)。当process_payment任务成功完成时，其返回值（此处为order_id）将作为参数传递给send_notification_email任务执行。

4. delay()与apply_async()：何时选择何者？

现在，您应该清楚了delay()和apply_async()的区别及各自的优点。那么，实际在什么情况下该选择哪个方法呢？

  • delay()：

    • 当您希望以最简单的方式异步执行任务时
    • 当不需要对任务执行时机进行特别控制时
    • 当您重视代码的简洁性时

  • apply_async()：

    • 当需要将任务执行预定在特定时间之后或特定时刻时（countdown、eta）
    • 当希望设置任务的过期时间以避免不必要的执行时（expires）
    • 当希望将任务路由到特定队列以控制工作组时（queue）
    • 当需要管理任务的优先级时（priority）
    • 当希望明确指定任务的序列化或压缩方式时（serializer、compression）
    • 当希望在任务成功或失败时自动执行其他任务的回调时（link、link_error）

一般而言，对于简单的异步处理，使用delay()较为方便，而对于更为复杂和细致的控制，则建议使用apply_async()。delay()是apply_async()的简化形式，因此无论如何都可以使用apply_async()实现相同（或更多）的功能。

总结

今天，我们深入探讨了Celery强大工具apply_async()方法，了解了它与delay()的关系及各种应用方式。apply_async()使得管理异步任务更加精细，有效利用Celery满足服务需求。

在下一篇最后的文章中，我们将比较分析定义Celery任务的核心装饰器@app.task与@shared_task，助您选择与项目结构和开发理念相符合的任务定义方式。感谢您一路陪伴我们的Celery之旅！

未查看之前文章的朋友请查看以下链接！

# Celery的魔法_ 深入解析delay()的表面与本质