等待队列填充 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进程在填充队列之前启动，以便在填充队列时使用队列。 我认为有解决方法可以传达何时停止迭代，但我想知道是否存在正确的方法来做到这一点。 我尝试了两种多处理。管道和.锁定，但我不知道如何正确有效地实施。

已解决：这是最好的方法吗？

以下代码在 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方法，这在我的情况下是有限的。

Answer 1

您要求的是queue.get的默认行为。它将等待(阻止(直到队列中有可用的项目。发送哨兵值确实是结束子进程的首选方法。

您的方案可以简化为如下所示：

import random
import time
from multiprocessing import Manager, Process

def save_data(save_que, file_):
for data in iter(save_que.get, 'STOP'):
print("saving data", data)
print("All data saved")
return

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")

if __name__ == '__main__':
manager = Manager()
save_que = manager.Queue()
file_ = "file"
save_p = Process(target=save_data, args=(save_que, file_))
save_p.start()
produce_data(save_que)
save_p.join()

---

编辑以回答评论中的问题：

如果提示被多个不同的代理访问，并且每个代理都有完成其任务的随机时间，我应该如何实现停止消息？

这没有太大区别，您必须将尽可能多的哨兵值放入队列中，就像您拥有的消费者一样。

一个实用程序函数，它返回一个流记录器以查看操作的位置：

def get_stream_logger(level=logging.DEBUG):
"""Return logger with configured StreamHandler."""
stream_logger = logging.getLogger('stream_logger')
stream_logger.handlers = []
stream_logger.setLevel(level)
sh = logging.StreamHandler()
sh.setLevel(level)
fmt = '[%(asctime)s %(levelname)-8s %(processName)s] --- %(message)s'
formatter = logging.Formatter(fmt)
sh.setFormatter(formatter)
stream_logger.addHandler(sh)
return stream_logger

与多个使用者一起编写代码：

import random
import time
from multiprocessing import Manager, Process
import logging
def save_data(save_que, file_):
stream_logger = get_stream_logger()
for data in iter(save_que.get, 'STOP'):
time.sleep(random.randint(1, 5))  # random delay
stream_logger.debug(f"saving: {data}")  # DEBUG
stream_logger.debug("all data saved")  # DEBUG
return

def produce_data(save_que, n_workers):
stream_logger = get_stream_logger()
for _ in range(5):
time.sleep(random.randint(1, 5))
data = random.randint(1, 10)
stream_logger.debug(f"producing: {data}")  # DEBUG
save_que.put(data)
for _ in range(n_workers):
save_que.put("STOP")

if __name__ == '__main__':
file_ = "file"
n_processes = 2
manager = Manager()
save_que = manager.Queue()
processes = []
for _ in range(n_processes):
processes.append(Process(target=save_data, args=(save_que, file_)))
for p in processes:
p.start()
produce_data(save_que, n_workers=n_processes)
for p in processes:
p.join()

示例输出：

[2018-09-02 20:10:35,885 DEBUG    MainProcess] --- producing: 2
[2018-09-02 20:10:38,887 DEBUG    MainProcess] --- producing: 8
[2018-09-02 20:10:38,887 DEBUG    Process-2] --- saving: 2
[2018-09-02 20:10:39,889 DEBUG    MainProcess] --- producing: 8
[2018-09-02 20:10:40,889 DEBUG    Process-3] --- saving: 8
[2018-09-02 20:10:40,890 DEBUG    Process-2] --- saving: 8
[2018-09-02 20:10:42,890 DEBUG    MainProcess] --- producing: 1
[2018-09-02 20:10:43,891 DEBUG    Process-3] --- saving: 1
[2018-09-02 20:10:46,893 DEBUG    MainProcess] --- producing: 5
[2018-09-02 20:10:46,894 DEBUG    Process-3] --- all data saved
[2018-09-02 20:10:50,895 DEBUG    Process-2] --- saving: 5
[2018-09-02 20:10:50,896 DEBUG    Process-2] --- all data saved
Process finished with exit code 0

Answer 2

不完全相关，但下面的解决方案允许您等到队列中有某些内容后再对其执行操作。就我而言，我有一个线程，它等到数据被放入队列，然后假脱机一个进程

from multiprocessing import get_context
import multiprocessing.queues as mpq
import multiprocessing.connection as connection
from threading import Thread
class SpoolingQueue(mpq.Queue):
def __init__(self,*args,**kwargs):
ctx = get_context()
super(SpoolingQueue, self).__init__(*args, **kwargs, ctx=ctx)
def wait(self):
connection.wait([self._reader])

def task(q:SpoolingQueue):
while True:
q.wait()
print(q.get()) #Do something here knowing that the queue has data
if __name__ == '__main__':
import time
q=SpoolingQueue()
canary = Thread(target=task, args=(q,))
canary.start()
time.sleep(5)
q.put('Output')