用户启动的后台进程数量有限

Question

我需要允许用户为非常非常大的工作提交请求。我们说的是 100 GB 的内存和 20 小时的计算时间。这花费了我们公司很多钱，所以规定任何时候只能运行 2 个作业，并且当 2 个已经在运行时请求新作业将被拒绝（并且用户通知服务器忙）。

我当前的解决方案使用来自 concurrent.futures 的 Executor，并且需要将 Apache 服务器设置为仅运行一个进程，从而降低响应速度（当前用户数非常少，所以暂时还可以）。

如果可能的话，我想为此使用 Celery，但我在文档中没有看到任何方法可以完成此特定设置。

如何在 Django 应用程序的后台运行数量有限的作业，并在作业因服务器繁忙而被拒绝时通知用户？

Answer 1

对于这种特殊情况，我有两种解决方案，一种是 celery 开箱即用的解决方案，另一种是您自己实现的。

1. 你可以对芹菜工人做这样的事情。特别是，您只创建了两个并发=1的工作进程（或者，一个并发=2的工作进程，但这将是线程，而不是不同的进程），这样，只能完成两项工作异步。现在您需要一种在两个作业都被占用时引发异常的方法，然后使用inspect 来计算活动任务的数量并在需要时抛出异常。如需实施，您可以查看this SO post。

您可能还对rate limits 感兴趣。

1. 您可以使用选择的锁定解决方案自己完成所有操作。特别是，一个很好的实现可以确保只有两个进程使用 redis（和 redis-py）运行，如下所示。 （考虑到你知道redis，因为你知道celery）

   from redis import StrictRedis
   
   redis = StrictRedis('localhost', '6379')
   locks = ['compute:lock1', 'compute:lock2']
   for key in locks:
       lock = redis.lock(key, blocking_timeout=5)
       acquired = lock.acquire()
       if acquired:
           do_huge_computation()
           lock.release()
           break
       print("Gonna try next possible slot")
   
   if not acquired:
       raise SystemLimitsReached("Already at max capacity !")
   

这样可以确保系统中只能存在两个正在运行的进程。第三个进程将在lock.acquire() 行中阻塞 blocking_timeout 秒，如果锁定成功，acquired 将为 True，否则为 False，您会告诉用户等待！

过去某个时候我有同样的要求，我最终编写的代码类似于上面的解决方案。特别是

1. 这具有尽可能少的竞争条件
2. 易于阅读
3. 不依赖于系统管理员，突然将负载下工作人员的并发性加倍并炸毁整个系统。
4. 您还可以实现每个用户的限制，这意味着每个用户可以同时运行 2 个作业，只需更改 compute:lock1 到 compute:userId:lock1 和 lock2 相应地。你不能用香草芹菜来做这个。

Answer 2

第一

您需要SpiXel's solution 的第一部分。据他说，“你只创建了两个并发=1的工作进程”。

第二

为队列中等待的任务设置time out，根据how to limit number of tasks in queue and stop feeding when full?设置CELERY_EVENT_QUEUE_TTL和队列长度限制。

因此，当这两个工作正在运行作业，并且队列中的任务等待 10 秒或您喜欢的任何时间段时，该任务将超时。或者如果队列已经完成，新到达的任务将被丢弃。

第三

你需要额外的东西来处理通知“当作业因为服务器繁忙而被拒绝时通知用户”。

Dead Letter Exchanges 是您所需要的。每次任务因队列长度限制或消息超时而失败。 “一旦达到限制，消息将从队列的前面被丢弃或死信，以便为新消息腾出空间。”

您可以设置“x-dead-letter-exchange”路由到另一个队列，一旦这个队列收到死信消息，您可以向用户发送通知消息。

Answer 3

首先，您需要限制工作线程的并发性 (docs)：

celery -A proj worker --loglevel=INFO --concurrency=2 -n <worker_name>

这将有助于确保您的活动任务不超过 2 个，即使代码中出现错误也是如此。

现在您有 2 种方法来实现任务编号验证：

1. 您可以使用inspect 获取活动和计划任务的数量：

    from celery import current_app
   
    def start_job():
         inspect = current_app.control.inspect()
         active_tasks = inspect.active() or {}
         scheduled_tasks = inspect.scheduled() or {}
         worker_key = 'celery@%s' % <worker_name>
         worker_tasks = active_tasks.get(worker_key, []) + scheduled_tasks.get(worker_key, [])
         if len(worker_tasks) >= 2:
             raise MyCustomException('It is impossible to start more than 2 tasks.') 
         else:
             my_task.delay()
   

2. 您可以将当前正在执行的任务数存储在数据库中，并据此验证任务执行情况。

如果您想扩展您的功能，第二种方法可能会更好 - 引入高级用户或不允许一个用户执行 2 个请求。