単純な並列処理②生産に時間がかかる時 (Python; multiprocessing.Queue)

概要

多くの重い処理から分担して生産するための簡潔な実装をする。

multicoreを活用したいので、multiprocessingを用いる。

内容

環境

macOS Catalina
Python 3.7.0

準備

from multiprocessing import Process, Queue
import time
import sys
import random

def reader_proc(queue):
    done = 0
    while True:
        msg = queue.get()
        if (msg == 'DONE'):
            done += 1
        if done == 2:
            break

def writer_proc(name, queue):
    for i in range(3):
        queue.put(i)
        print(name, i)
        time.sleep(random.randint(0, 5))
    queue.put('DONE')
    print(name, 'DONE')

queue = Queue()
reader = Process(target=reader_proc, args=(queue,))
writer1 = Process(target=writer_proc, args=('write1', queue))
writer2 = Process(target=writer_proc, args=('write2', queue))

reader_proc：queueに溜まったものを迅速に消費してくれる。
writer_proc：queueに追加してくれるが、生産するのに少し時間がかかる。

実行

reader.start()
writer1.start()
writer2.start()

reader.join()
writer1.join()
writer2.join()

output

write1 0
write2 0
write2 1
write2 2
write1 1
write2 DONE
write1 2
write1 DONE

蛇足

各所の説明では、いつ終わるか分からないQueueに未対応のものが多かった。
こうすれば、好きなタイミングで綺麗に終わらせられる。

'DONE'が重要なデータでないか不安な場合は、以前の記事を参照すればすぐに解決する。

参考にさせていただいた本・頁

• https://stackoverflow.com/questions/11515944/how-to-use-multiprocessing-queue-in-python
• https://qiita.com/yo314159265/items/0215bb1f9128800fc7f0 (iteratorシグナル記事)
• https://qiita.com/yo314159265/items/dbedf0bb27589af6a33d (前回記事)

感想

たまに使うので、整理できて良かった。
reader_procに処理の個数を入れているのはあまり美しくはないが、行数優先。

今後

このシリーズはもう少し続けます。

追記：続き書きました。(https://qiita.com/yo314159265/items/00ac562a5ae777364e6b)