在Python中优化多处理（后续：使用队列）

Question

这是this的后续问题。用户将建议使用队列，我尝试实现下面的解决方案。当j=1000时，这个解决方案运行得很好，但是，当我尝试扩展到更大的数字时，它挂起了。我被困在这里，无法确定它为什么会挂起来。如有任何建议，将不胜感激。而且，代码开始变得难看，因为我一直在弄乱它，我为所有嵌套函数道歉。你知道吗

def run4(j):
    """
    a multicore approach using queues
    """
    from multiprocessing import Process, Queue, cpu_count
    import os

    def bazinga(uncrunched_queue, crunched_queue):
        """
        Pulls the next item off queue, generates its collatz
        length and 
        """
        num = uncrunched_queue.get()
        while num != 'STOP': #Signal that there are no more numbers
            length = len(generateChain(num, []) )
            crunched_queue.put([num , length])
            num = uncrunched_queue.get()

    def consumer(crunched_queue):
        """
        A process to pull data off the queue and evaluate it
        """
        maxChain = 0
        biggest = 0
        while not crunched_queue.empty():
            a, b = crunched_queue.get()
            if b > maxChain:
                biggest = a
                maxChain = b
        print('%d has a chain of length %d' % (biggest, maxChain))

    uncrunched_queue = Queue()
    crunched_queue = Queue()
    numProcs = cpu_count()

    for i in range(1, j): #Load up the queue with our numbers
        uncrunched_queue.put(i)
    for i in range(numProcs): #put sufficient stops at the end of the queue
        uncrunched_queue.put('STOP')

    ps = []
    for i in range(numProcs):
        p = Process(target=bazinga, args=(uncrunched_queue, crunched_queue))
        p.start()
        ps.append(p)

    p = Process(target=consumer, args=(crunched_queue, ))
    p.start()
    ps.append(p)

    for p in ps: p.join()

Answer 1

您正在将'STOP'毒丸放入uncrunched_queue（您应该这样做），并相应地关闭生产商；另一方面，您的消费者只检查已处理队列的空值：

while not crunched_queue.empty():

（这项工作完全取决于比赛条件，顺便说一句，这是不好的）

当您开始向bazinga生产者抛出非琐碎的工作单元时，它们需要更长的时间。如果所有这些花费足够长的时间，你的crunched_queue就会干涸，你的消费者就会死亡。我认为你可能误认了发生了什么-你的程序没有“挂起”，它只是停止输出的东西，因为你的消费者已经死了。你知道吗

你需要实施一个更聪明的方法来关闭你的消费者。要么寻找n毒丸，其中n是生产商的数量（相应地，当他们关闭时，每个生产商在crunched_queue中扔一个），要么使用Semaphore之类的东西，对每个活的生产商计数，当一个生产商关闭时计数。你知道吗