等待队列填充python多处理的最佳方法

Question

这是我第一次认真对待并行计算。我在python中使用multiprocessing模块，我遇到了这个问题：

队列使用者在不同的进程中运行，然后运行队列生成器，前者应该等待后者完成其作业，然后再停止迭代队列。有时消费者比生产者更快，队列保持空白。如果我没有提出任何条件，程序将不会停止。

在示例代码中，我使用通配符PRODUCER_IS_OVER来举例说明我需要的内容。

以下代码草绘问题：

def save_data(save_que, file_):
    ### Coroutine instantiation
    PRODUCER_IS_OVER = False
    empty = False
    ### Queue consumer
    while not(empty and PRODUCER_IS_OVER):
        try:
            data = save_que.get()
            print("saving data",data)
        except:
            empty = save_que.empty()
            print(empty)
            pass
        #PRODUCER_IS_OVER = get_condition()
    print ("All data saved")
    return

def get_condition():
    ###NameError: global name 'PRODUCER_IS_OVER' is not defined
    if PRODUCER_IS_OVER:
        return True
    else:
        return False


def produce_data(save_que):
    for _ in range(5):
        time.sleep(random.randint(1,5))
        data = random.randint(1,10)
        print("sending data", data)
        save_que.put(data)

### Main function here
import random
import time
from multiprocessing import Queue, Manager, Process
manager = Manager()
save_que = manager.Queue()
file_ = "file"
save_p    = Process(target= save_data, args=(save_que, file_))
save_p.start()
PRODUCER_IS_OVER = False
produce_data(save_que)
PRODUCER_IS_OVER = True
save_p.join()

produce_data需要可变时间，我希望在填充队列之前启动save_p进程，以便在填充时使用队列。我认为有办法解决何时停止迭代，但我想知道是否存在一种正确的方法来实现它。我尝试了multiprocessing.Pipe和.Lock，但我不知道如何正确有效地实现。

解决：这是最好的方式吗？

下面的代码在Q中实现STOPMESSAGE，工作正常，我可以用类QMsg对它进行优化，以防语言只支持静态类型。

def save_data(save_que, file_):
    # Coroutine instantiation
    PRODUCER_IS_OVER = False
    empty = False
    # Queue consumer
    while not(empty and PRODUCER_IS_OVER):
        data = save_que.get()
        empty = save_que.empty()
        print("saving data", data)
        if data == "STOP":
            PRODUCER_IS_OVER = True
    print("All data saved")
    return


def get_condition():
    # NameError: global name 'PRODUCER_IS_OVER' is not defined
    if PRODUCER_IS_OVER:
        return True
    else:
        return False


def produce_data(save_que):
    for _ in range(5):
        time.sleep(random.randint(1, 5))
        data = random.randint(1, 10)
        print("sending data", data)
        save_que.put(data)
    save_que.put("STOP")


# Main function here
import random
import time
from multiprocessing import Queue, Manager, Process
manager = Manager()
save_que = manager.Queue()
file_ = "file"
save_p = Process(target=save_data, args=(save_que, file_))
save_p.start()
PRODUCER_IS_OVER = False
produce_data(save_que)
PRODUCER_IS_OVER = True
save_p.join()

但是，如果队列是由几个独立的进程生成的，那么这不起作用：在这种情况下谁将发送ALT消息？

另一种解决方案是将进程索引存储在列表中并执行：

def some_alive():
    for p in processes:
        if p.is_alive():
            return True
    return False

但是multiprocessing仅在父进程中支持.is_alive方法，这在我的情况下是有限的。

谢谢