Совместное использование нескольких очередей среди процессов в Python

Похоже, что ваши проблемы начались, когда вы пытались передать multiprocessing.Queue() , передав его в качестве аргумента. Вы можете обойти это, создав вместо этого управляемую очередь :

 import multiprocessing manager = mutiprocessing.Manager() passable_queue = manager.Queue() 

Когда вы используете менеджера для его создания, вы сохраняете и передаете прокси-сервер в очередь, а не в очередь, так что даже когда объект, который вы передаете своим рабочим процессам, скопирован, он все равно будет указывать на то же самое структура данных: ваша очередь. Он очень похож (по понятию) на указатели на C / C ++. Если вы создадите свои очереди таким образом, вы сможете передать их при запуске рабочего процесса.

Поскольку теперь вы можете передавать очереди, вам больше не нужен ваш словарь для управления. Держите нормальный словарь в главном, который сохранит все сопоставления и только даст вашему работнику обрабатывать очереди, в которых они нуждаются, поэтому им не нужен доступ к каким-либо сопоставлениям.

Я написал пример этого здесь. Похоже, вы проходите объекты между вашими работниками, вот что здесь делается. Представьте, что у нас есть две стадии обработки, и данные начинаются и заканчиваются в контроле main . Посмотрите, как мы можем создавать очереди, которые соединяют рабочих как конвейер, но, предоставляя им только очереди, которые им нужны , им не нужно знать о каких-либо сопоставлениях:

 import multiprocessing as mp def stage1(q_in, q_out): q_out.put(q_in.get()+"Stage 1 did some work.\n") return def stage2(q_in, q_out): q_out.put(q_in.get()+"Stage 2 did some work.\n") return def main(): pool = mp.Pool() manager = mp.Manager() # create managed queues q_main_to_s1 = manager.Queue() q_s1_to_s2 = manager.Queue() q_s2_to_main = manager.Queue() # launch workers, passing them the queues they need results_s1 = pool.apply_async(stage1, (q_main_to_s1, q_s1_to_s2)) results_s2 = pool.apply_async(stage2, (q_s1_to_s2, q_s2_to_main)) # Send a message into the pipeline q_main_to_s1.put("Main started the job.\n") # Wait for work to complete print(q_s2_to_main.get()+"Main finished the job.") pool.close() pool.join() return if __name__ == "__main__": main() 

Код выводит этот результат:

Главное начало работу.
Этап 1 проделал определенную работу.
Этап 2 сделал некоторую работу.
Главная закончила работу.

Я не включил пример хранения очередей или объектов AsyncResults в словарях, потому что я до сих пор не совсем понимаю, как должна работать ваша программа. Но теперь, когда вы можете свободно передавать свои очереди, вы можете построить свой словарь для хранения сопоставлений очереди / процессов по мере необходимости.

На самом деле, если вы действительно строите конвейер между несколькими рабочими, вам даже не нужно указывать ссылки на «межличностные» очереди в main . Создайте очереди, передайте их своим работникам, а затем сохраните ссылки на очереди, которые будут использоваться main . Я определенно рекомендовал бы ставить старую очередь на сбор мусора как можно быстрее, если у вас действительно есть «произвольное число» очередей.