все Python 3multiprocessingпонятие

multiprocessing.JoinableQueue

обмен данными

Очередь процессов с учётом выполненных задач: потребитель зовёт task_done() на каждый get(), а join() ждёт, пока все положенные задачи не будут отмечены.

Синтаксис

q = ctx.JoinableQueue(); q.put(x); q.get(); q.task_done(); q.join()

Забытый task_done() — и q.join() висит вечно, причём без всякой ошибки; лишний вызов даёт ValueError. task_done() зовут и для «ядовитой» метки остановки (None), иначе счётчик не сойдётся. Как и у Queue, потребитель обычно бесконечный цикл, поэтому после q.join() его нужно остановить — меткой или daemon=True.

Примеры

import multiprocessing as mp
def consumer(q, out):
    while True:
        item = q.get()
        if item is None:
            q.task_done()
            break
        out.put(item * 10)
        q.task_done()
if __name__ == '__main__':
    ctx = mp.get_context('spawn')
    q, out = ctx.JoinableQueue(), ctx.Queue()
    p = ctx.Process(target=consumer, args=(q, out))
    p.start()
    for item in (1, 2, 3, None):
        q.put(item)
    q.join()  # ждёт task_done() на каждый put()
    print(sorted(out.get(timeout=10) for _ in range(3)))  # → [10, 20, 30]
    p.join()

См. также

Открыть в полном глоссарииdocs.python.orgСообщить о неточности