使用进程的 Python 多处理：消耗大量内存

Question

我正在从单个 python 代码运行多个进程：

代码片段：

while 1:
   if sqsObject.msgCount() > 0:
        ReadyMsg = sqsObject.readM2Q()
        if ReadyMsg == 0:
            continue
        fileName = ReadyMsg['fileName']
        dirName  = ReadyMsg['dirName']
        uuid         = ReadyMsg['uid']
        guid         = ReadyMsg['guid']
        callback     = ReadyMsg['callbackurl']

        # print ("Trigger Algorithm Process")
        if(countProcess < maxProcess):

           try:
             retValue = Process(target=dosomething, args=(dirName, uuid,guid,callback))
             processArray.append(retValue)
             retValue.start()
             countProcess = countProcess + 1
           except:
             print "Cannot Run Process"
        else:
           for i in range(len(processArray)):
              if (processArray[i].is_alive() == True):
                 continue
              else:
                 try:
                    #print 'Restart Process'
                    processArray[i] = Process(target=dosomething, args=(dirName,uuid,guid,callback))
                    processArray[i].start()
                 except:
                    print "Cannot Run Process"


   else: # No more request to service

       for i in range(len(processArray)):
            if (processArray[i].is_alive() == True):
                processRunning = 1
                break
            else:
                continue

      if processRunning == 0:
           countProcess = 0

      else:
           processRunning = 0

在这里，我正在从队列中读取消息并创建一个进程来对该消息运行算法。我设置了 maxProcess 的上限。因此，在达到 maxProcess 后，我想通过检查 is_alive() 来重用不活跃的 processArray 插槽。

此过程对于较少数量的进程运行良好，但是对于大量消息（例如 100 条），内存消耗会超出上限。我认为我通过重用进程槽而泄漏了。

不确定过程中出了什么问题。

提前感谢您发现错误或明智的建议。

Answer 1

总之，你的代码很奇怪:-)

不是an mvce，所以没有其他人可以测试它，但是看看它，你在内部循环中有这个（稍微简化的）结构：

if count < limit:
    ... start a new process, and increment count ...
else:
    do things that can potentially start even more processes
    (but never, ever, decrease count)

这似乎是不明智的。

在任何地方都没有调用流程实例的join()。 （稍后我们将回到外循环及其else 案例。）

让我们更仔细地看一下内部循环的else 案例代码：

   for i in range(len(processArray)):
        if (processArray[i].is_alive() == True):

抛开不必要的== True 测试——这有点冒险，因为is_alive() 方法并没有明确承诺返回True 和False，只是一些布尔值有效的东西——考虑@ 987654322@（此链接转到 py2k 文档，但 py3k 是相同的，您的 print 语句暗示您的代码无论如何都是 py2k）：

is_alive()

返回进程是否存活。

大致来说，一个进程对象从start() 方法返回的那一刻起一直处于活动状态，直到子进程终止。

由于我们看不到dosomething 的代码，因此很难说这些事情是否会终止。可能他们会这样做（通过退出），但如果他们不这样做，或者做得不够快，我们可能会在这里遇到问题，我们只是将我们从外部循环中拉出队列的消息丢弃。

如果他们做终止，我们只需从数组中删除进程引用，通过覆盖它：

            processArray[i] = Process(...)

processArray[i] 中的前一个值被丢弃。目前尚不清楚您是否已将其保存在其他任何地方，但如果您没有保存，则 Process 实例将被丢弃，现在实际上不可能调用它的 join() 方法。

一些 Python 数据结构在被放弃时倾向于自行清理（例如，打开的流刷新输出并根据需要关闭），但多进程代码似乎不会自动加入（）其子级。所以这可能是问题的根源。

最后，每当我们在外循环中找到else 的情况时，我们都会对任何活着的进程进行同样有点奇怪的搜索——顺便说一下，可以更清楚地写成：

if any(p.is_alive() for p in processArray):

只要我们不关心哪些特定的是活着的，哪些不是——如果没有人报告自己是活着的，我们会重置计数，但永远不要对变量做任何事情processArray，这样每个processArray[i] 仍然持有Process 实例的标识。 （所以至少我们可以在每一个上调用join，不包括因覆盖而丢失的任何内容。）

与其自己构建自己的Pool，不如使用multiprocess.Pool 及其apply 和apply_async 方法，如miraculixx's answer。

Answer 2

不确定过程中出了什么问题。

看起来您正在创建与消息一样多的进程，即使达到 maxProcess 计数也是如此。

我认为我通过重用进程槽而泄漏了。

无需自己管理流程。只需使用process pool：

 # before your while loop starts
 from multiprocessing import Pool
 pool = Pool(processes=max_process)
 while 1:
   ...
   # instead of creating a new Process
   res = pool.apply_async(dosomething, 
                          args=(dirName,uuid,guid,callback)) 
 # after the while loop has finished
 # -- wait to finish
 pool.close()
 pool.join()

提交作业的方式

请注意，Pool class 支持多种提交作业的方式：

• apply_async - 一次一条消息
• map_async - 一次一大块消息

如果消息到达的速度足够快，最好收集其中的几个（比如一次 10 或 100 个，具体取决于完成的实际处理）并使用 map 向目标函数提交“小批量”一次：

...
while True:
    messages = []
    # build mini-batch of messages
    while len(messages) < batch_size:
        ... # get message
        messages.append((dirName,uuid,guid,callback))
    pool.map_async(dosomething, messages)

为避免dosomething 留下的内存泄漏，您可以要求池在消耗一定数量的消息后重新启动进程：

max_tasks = 5 # some sensible number
Pool(max_processes, maxtasksperchild=max_tasks)

分布式

如果使用这种方法仍然超出内存容量，请考虑使用分布式方法，即添加更多机器。使用Celery，这将非常简单，来自上面：

# tasks.py
@task
def dosomething(...):
   ... # same code as before

# driver.py
  while True:
     ... # get messages as before
     res = somefunc.apply_async(args=(dirName,uuid,guid,callback))