Python多处理内存增加

Question

我有一个应该永远运行的程序。 这是我正在做的事情：

from myfuncs import do, process

class Worker(multiprocessing.Process):

    def __init__(self, lock):
        multiprocesing.Process.__init__(self)
        self.lock = lock
        self.queue = Redis(..) # this is a redis based queue
        self.res_queue = Redis(...)

     def run():
         while True:
             job = self.queue.get(block=True)
             job.results = process(job)
             with self.lock:
                 post_process(self.res_queue, job)


def main():
    lock = multiprocessing.Semaphore(1)
    ps = [Worker(lock) for _ in xrange(4)]
    [p.start() for p in ps]
    [p.join() for p in ps]

self.queue 和 self.res_queue 是两个像 python stdlib Queue 一样工作的对象，但它们 使用 Redis 数据库作为后端。

函数进程对job携带的数据（主要是html解析）和返回的数据做一些处理 一本字典。

函数 post_process 通过检查某些条件将作业写入另一个 redis 队列（一次只有一个进程可以检查导致锁定的条件）。它返回真/假。

程序每天使用的内存在增加。 有人能弄清楚发生了什么吗？

当作业超出运行方法的范围时，内存应该是空闲的吗？

Answer 1

当作业超出运行方法的范围时，内存应该是空闲的吗？

首先，作用域是整个run 方法，它永远循环，因此永远不会发生。 （此外，当您退出 run 方法时，进程将关闭并且它的内存无论如何都会被释放......）

但即使它确实超出了范围，也并不意味着您似乎认为它意味着什么。 Python 不像 C++，其中的变量存储在堆栈上。所有对象都存在于堆中，并且它们一直存在，直到不再有对它们的引用。超出范围的变量意味着该变量不再引用它曾经引用的任何对象。如果该变量是对该对象的唯一引用，那么它将被释放*，但如果您在其他地方进行了其他引用，则在这些其他引用消失之前无法释放该对象。

同时，超出范围并没有什么神奇之处。变量停止引用对象的任何方式都具有相同的效果——无论是变量超出范围，您在其上调用del，还是为它分配一个新值。因此，每次通过循环时，当您执行job = 时，您将删除先前对job 的引用，即使没有超出范围。 （但请记住，在高峰期您​​将有 两个 个工作，而不是一个，因为新工作在旧工作发布之前就已从队列中移除。如果这是一个问题，您总是可以这样做job = None 在队列中阻塞之前。）

所以，假设问题实际上是 job 对象（或它拥有的东西），问题是您没有向我们展示的一些代码在某处保留了对它的引用。

在不知道自己在做什么的情况下，很难提出修复建议。它可能只是“不要在那里存储”。或者它可能是“存储弱引用而不是对象本身”。或“添加 LRU 算法”。或者“添加一些流量控制，这样如果你备份太多，你就不会继续工作，直到内存用完”。

---

_{* 在 CPython 中，这会立即发生，因为垃圾收集器是基于引用计数的。另一方面，在 Jython 和 IronPython 中，垃圾收集器仅依赖于底层 VM 的垃圾收集器，因此在 JVM 或 CLR 注意到它不再被引用之前不会释放对象，这通常不是立即的，并且是不确定的.}

Answer 2

如果您找不到泄漏源，您可以通过让每个工作人员只处理有限数量的任务来解决它。一旦它们达到任务限制，您就可以让它们退出，并用新的工作进程替换它们。内置的 multiprocessing.Pool 对象通过 maxtasksperchild 关键字参数支持这一点。你可以做类似的事情：

import multiprocessing
import threading

class WorkerPool(object):
    def __init__(self, workers=multiprocessing.cpu_count(),
                 maxtasksperchild=None, lock=multiprocessing.Semaphore(1)):
        self._lock = multiprocessing.Semaphore(1)
        self._max_tasks = maxtasksperchild
        self._workers = workers
        self._pool = []
        self._repopulate_pool()
        self._pool_monitor = threading.Thread(self._monitor_pool)
        self._pool_monitor.daemon = True
        self._pool_monitor.start()

    def _monitor_pool(self):
        """ This runs in its own thread and monitors the pool. """
        while True:
            self._maintain_pool()
            time.sleep(0.1)

    def _maintain_pool(self):
        """ If any workers have exited, start a new one in its place. """
        if self._join_exited_workers():
            self._repopulate_pool()

    def _join_exited_workers(self):
        """ Find exited workers and join them. """
        cleaned = False
        for i in reversed(range(len(self._pool))):
            worker = self._pool[i]
            if worker.exitcode is not None:
                # worker exited
                worker.join()
                cleaned = True
                del self._pool[i]
        return cleaned

    def _repopulate_pool(self):
        """ Start new workers if any have exited. """
        for i in range(self._workers - len(self._pool)):
            w = Worker(self._lock, self._max_tasks)
            self._pool.append(w)
            w.start()    


class Worker(multiprocessing.Process):

    def __init__(self, lock, max_tasks):
        multiprocesing.Process.__init__(self)
        self.lock = lock
        self.queue = Redis(..) # this is a redis based queue
        self.res_queue = Redis(...)
        self.max_tasks = max_tasks

     def run():
         runs = 0
         while self.max_tasks and runs < self.max_tasks:
             job = self.queue.get(block=True)
             job.results = process(job)
             with self.lock:
                 post_process(self.res_queue, job)
            if self.max_tasks:
                 runs += 1


def main():
    pool = WorkerPool(workers=4, maxtasksperchild=1000)
    # The program will block here since none of the workers are daemons.
    # It's not clear how/when you want to shut things down, but the Pool
    # can be enhanced to support that pretty easily.

请注意，上面的池监控代码与 multiprocessing.Pool 中用于相同目的的代码几乎完全相同。