2

我正在尝试使用多处理和队列来实现生产者-消费者场景；主进程是生产者，两个子进程从队列中消费数据。这在没有任何异常情况发生的情况下有效，但问题是我希望能够重新启动工作人员以防他们死亡（kill -9 workerpid）。但是，当我杀死一个或两个工人时，即使主进程一直在队列中填充数据，他们也会开始说“队列为空”。

我在这里想念什么？（在 Ubuntu 12.04 上使用 Python 2.7.3）

import sys
import time

import multiprocessing
from Queue import Empty

workers = []
fqueue = multiprocessing.Queue()

class Worker(multiprocessing.Process):
    def run(self):
        queue = self._args[0]
        print "{0} starting up, queue at: {1}".format(self.pid, queue)
        while True:
            try:
                obj = queue.get(block=True, timeout=1)
                print "{0}: got from queue: {1}".format(self.pid, obj)
            except Empty:
                print "{0}: queue was empty".format(self.pid)
                continue
            except IOError, e:
                print "{0}: got IOError on queue: {1}".format(self.pid, e)
                return

if __name__ == "__main__":
    print "zipper starting up (2 workers)"
    for _ in range(0, 2):
        p = Worker(args=(fqueue,)) 
        workers.append(p)
        p.start()

    cnt = 0
    while True:
        for i in range(0, len(workers)):
            p = workers[i]
            if not p.is_alive():
                print "main: worker {0} is not alive".format(p.pid)
                p = Worker(args=(fqueue,))
                print "main: restarted worker: {0}".format(p)
                p.start()
                workers[i] = p
        print "main: tick"
        cnt += 1
        fqueue.put(cnt)
        time.sleep(2)
4

1 回答 1

5

您是否看到文档中的警告？：

警告

如果一个进程在尝试使用队列时被 Process.terminate() 或 os.kill() 杀死，那么队列中的数据可能会损坏。这可能会导致任何其他进程在稍后尝试使用队列时出现异常。

所以杀死一个正在使用队列的进程可以使整个队列不可用。

于 2012-05-29T08:50:24.543 回答