python - 管理具有不同内存使用情况的 Python 多进程进程

Question

我使用一个简单的 RabbitMQ 队列将任务分配给工作进程。每个工作进程使用一个多处理实例池来同时处理多个任务，以尽可能多地使用内存和CPU。

问题是，某些任务比其他任务占用更多的 RAM，因此如果启动多个实例，工作进程就会崩溃。但是，当工作线程正在处理 RAM 密集型任务时，我希望它能够处理其他 RAM 密集型任务，以使用其余的 CPU。

一个想法是使用多个队列或主题，但我想知道推荐的方法是什么。我可以在进程崩溃之前捕获内存不足错误吗？

解决这个问题的正确方法是什么？

[更新更新]

整个系统将由多个多核机器组成，但在每个多核机器上只有一个运行的工作程序，它创建与核心一样多的多处理实例。不同的机器应该彼此独立，只是它们从同一个队列获取任务。

Answer 1

我认为 try catch OOM 错误并从中恢复将是非常困难的，甚至是不可能的。您需要运行一个线程或进程来不断监视内存使用情况，当它检测到内存使用率过高时，......到底是什么？杀死正在处理任务的进程？尝试暂停它(如果可能的话；它可能不取决于您的任务正在做什么)。即使这样，暂停也不会释放任何内存。您必须释放内存并在安全时重新启动任务，这意味着您必须重新排队，决定何时安全，等等。

我建议不要尝试检测问题并从中恢复，而是尝试完全避免它。创建两个队列和两个池。一个队列/池用于高内存任务，另一个队列/池用于低内存任务。高内存池中只有一个进程，因此它仅限于同时运行一个任务，这可以节省内存。低内存队列将具有 multiprocessing.cpu_count() - 1 进程，允许您在两个池中保持 CPU 饱和。

这种方法的一个潜在问题是，如果您耗尽了高内存队列，同时仍有低内存任务待处理，那么您将浪费一个 CPU。您可以以非阻塞方式(或超时)处理高内存队列的这种消耗，这样，当您准备好消耗任务时，如果高内存队列为空，您可以获取低内存队列相反，内存任务。然后，当处理完成后，再次检查高内存队列。

类似这样的事情:

import multiprocessing

# hi_q and lo_q are placeholders for whatever library you're using to consume from RabbitMQ

def high_mem_consume():
    while True:
       task = hi_q.consume(timeout=2)
       if not task:
          lo_q.consume(timeout=2)
       if task:
           process_task(task)


def low_mem_consume():
    while True:
        task = lo_q.consume()  # Blocks forever
        process_task(task)


if __name__ == "__main__":

    hi_pool = multiprocessing.Pool(1)
    lo_pool = multiprocessing.Pool(multiprocessing.cpu_count() - 1)
    hi_pool.apply_async(high_mem_consume)
    lo_pool.apply_async(lo_mem_consume)