Совместное использование очереди результатов между несколькими процессами

В документации к multiprocessingмодулю показано, как передать очередь процессу, запущенному с multiprocessing.Process. Но как я могу разделить очередь с запущенными асинхронными рабочими процессами apply_async? Мне не нужно динамическое объединение или что-то еще, просто способ для рабочих (многократно) сообщать о своих результатах обратно на базу.

import multiprocessing
def worker(name, que):
    que.put("%d is done" % name)

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=3)
    q = multiprocessing.Queue()
    workers = pool.apply_async(worker, (33, q))

Это терпит неудачу с: RuntimeError: Queue objects should only be shared between processes through inheritance. Я понимаю, что это означает, и я понимаю совет наследовать, а не требовать травления / удаления (и все специальные ограничения Windows). Но как же я прохожу очереди таким образом , что работает? Я не могу найти пример и пробовал несколько альтернатив, которые по разным причинам не увенчались успехом. Помогите, пожалуйста?

Попробуйте использовать multiprocessing.Manager для управления очередью, а также сделать ее доступной для разных рабочих.

import multiprocessing
def worker(name, que):
    que.put("%d is done" % name)

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=3)
    m = multiprocessing.Manager()
    q = m.Queue()
    workers = pool.apply_async(worker, (33, q))

multiprocessing.Poolуже имеет общую очередь результатов, нет необходимости дополнительно задействовать Manager.Queue. Manager.Queueпредставляет собой queue.Queue(многопоточную очередь) под капотом, расположенную на отдельном серверном процессе и доступную через прокси. Это добавляет дополнительные накладные расходы по сравнению с внутренней очередью Pool. В отличие от использования встроенной обработки результатов Pool, порядок результатов в Manager.Queueтакже не гарантируется.

Рабочие процессы не запускаются .apply_async(), это уже происходит при создании экземпляра Pool. Что это началось , когда вы звоните pool.apply_async()новая «работа». Рабочие процессы пула запускают multiprocessing.pool.workerфункцию под капотом. Эта функция заботится об обработке новых «задач», передаваемых через внутренний пул, Pool._inqueueи об отправке результатов обратно родительскому объекту через Pool._outqueue. Указанное funcвами будет выполнено в пределах multiprocessing.pool.worker. funcнужно только returnчто-то, и результат будет автоматически отправлен обратно родителю.

.apply_async() немедленно (асинхронно) возвращает AsyncResultобъект (псевдоним для ApplyResult). Вам нужно вызвать .get()(блокирует) этот объект, чтобы получить фактический результат. Другой вариант - зарегистрировать функцию обратного вызова , которая запускается, как только становится готов результат.

from multiprocessing import Pool

def busy_foo(i):
    """Dummy function simulating cpu-bound work."""
    for _ in range(int(10e6)):  # do stuff
        pass
    return i

if __name__ == '__main__':

    with Pool(4) as pool:
        print(pool._outqueue)  # DEMO
        results = [pool.apply_async(busy_foo, (i,)) for i in range(10)]
        # `.apply_async()` immediately returns AsyncResult (ApplyResult) object
        print(results[0])  # DEMO
        results = [res.get() for res in results]
        print(f'result: {results}')       

Пример вывода:

<multiprocessing.queues.SimpleQueue object at 0x7fa124fd67f0>
<multiprocessing.pool.ApplyResult object at 0x7fa12586da20>
result: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

Примечание. Указание параметра timeout-параметра для .get()не остановит фактическую обработку задачи внутри рабочего процесса, а только разблокирует ожидающего родителя, подняв multiprocessing.TimeoutError.