有没有办法在芹菜中定义任务配额？答案

【问题标题】：Is there any way to define task quota in celery?有没有办法在芹菜中定义任务配额？
【发布时间】：2015-09-30 05:50:11
【问题描述】：

我有要求：

我几乎没有消耗大量资源的任务 - 导出需要大型复杂查询、子查询的不同报告
有很多用户。
我在 django 中构建了项目，并使用 celery 对任务进行了排队
我想限制用户，以便他们每分钟可以请求 10 个报告。这个想法是他们可以在 10 分钟内发出数百个请求，但我希望 celery 为用户执行 10 个任务。这样每个用户都可以轮到他们。

有什么办法可以让芹菜做到这一点？

谢谢

【问题讨论】：

【解决方案1】：

Celery 有一个设置来控制 RATE_LIMIT (http://celery.readthedocs.org/en/latest/userguide/tasks.html#Task.rate_limit)，这意味着可以在一个时间范围内运行的任务数。您可以将其设置为“100/m”（每秒一百个），以使您的系统每秒允许 100 个任务，重要的是要注意，该设置不是每个用户也不是任务，而是每个时间范围。您是否考虑过这种方法而不是限制每个用户？

为了让每个任务和用户对有一个“rate_limit”，您必须这样做。我认为（不确定）您可以根据需要使用 TaskRouter 或信号。 TaskRouters (http://celery.readthedocs.org/en/latest/userguide/routing.html#routers) 允许将任务路由到应用某些逻辑的指定队列。信号 (http://celery.readthedocs.org/en/latest/userguide/signals.html) 允许在任务调度周期的几个明确定义的点执行代码。

Router 的逻辑示例如下：

if task == 'A':
    user_id = args[0]  # in this task the user_id is the first arg
    qty = get_task_qty('A', user_id)
    if qty > LIMIT_FOR_A:
        return
elif task == 'B':
    user_id = args[2]  # in this task the user_id is the seconds arg
    qty = get_task_qty('B', user_id)
    if qty > LIMIT_FOR_B:
        return
return {'queue': 'default'}

使用上述方法，每次启动任务时，您应该在某个地方（例如 Redis）将 user_id/task_type 对和每次任务完成时，您都应该在同一个地方减少该值。

它似乎有点复杂，难以维护，对我来说几乎没有故障点。

我认为可能适合的其他方法是为每个用户和任务实现某种“分布式信号量”（类似于分布式锁），因此在每个需要限制运行任务数量的任务中，您可以使用它.

这个想法是，每次启动一个应该具有“并发控制”的任务时，它必须检查是否有一些资源可用，如果不只是返回。

你可以想象这个想法如下：

@shared_task
def my_task_A(user_id, arg1, arg2):
    resource_key = 'my_task_A_{}'.format(user_id)
    available = SemaphoreManager.is_available_resource(resource_key)
    if not available:
        # no resources then abort
        return

    try:
        # the resourse could be acquired just before us for other
        if SemaphoreManager.acquire(resource_key):
            #execute your code
    finally:
        SemaphoreManager.release(resource_key)

很难说您应该采用哪种方法，因为这取决于您的应用程序。

希望对你有帮助！

祝你好运！

【讨论】：