プロセスやキューを使用すると、いくつか問題が発生します。
次のコードを実行すると、ターゲット関数は単純にマスターキューから項目を取得し、それをそのプロセスに固有の別のキューに追加します。
import sys
import multiprocessing
from Queue import Empty
# This is just taking a number from the queue
# and adding it to another queue
def my_callable(from_queue, to_queue):
while True:
try:
tmp = from_queue.get(0)
to_queue.put(tmp)
print to_queue
except Empty:
break
# Create a master queue and fill it with numbers
main_queue = multiprocessing.Queue()
for i in xrange(100):
main_queue.put(i)
all_queues = []
processes = []
# Create processes
for i in xrange(5):
# Each process gets a queue that it will put numbers into
queue = multiprocessing.Queue()
# Keep up with the queue we are creating so we can get it later
all_queues.append(queue)
# Pass in our master queue and the queue we are transferring data to
process = multiprocessing.Process(target=my_callable,
args=(main_queue, queue))
# Keep up with the processes
processes.append(process)
for thread in processes:
thread.start()
for thread in processes:
thread.join()
ターゲット関数が使用中のキューを出力すると、1つのキューがほぼ排他的に使用されていることがわかります。
その後、出力を受け取って印刷すると、ほとんどの番号が1つのキューにまとめられていることがわかります。
def queue_get_all(q):
items = []
maxItemsToRetreive = 100
for numOfItemsRetrieved in range(0, maxItemsToRetreive):
try:
if numOfItemsRetrieved == maxItemsToRetreive:
break
items.append(q.get_nowait())
except Empty, e:
break
return items
for tmp in all_queues:
print queue_get_all(tmp)
何が原因ですか？自分のコードに、これらのプロセスが行っている作業を均等にするようなものがあるでしょうか。
出力
[0, 2, 3, 4, 5, 6, 7, 8]
[1, 9, 10]
[11, 14, 15, 16]
[12, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99]
[13]
回答：
回答№1は2私はあなたには2つの問題があると思います。
def my_callable(from_queue, to_queue):
while True:
try:
tmp = from_queue.get(0)
to_queue.put(tmp)
print to_queue
except Empty:
break
のドキュメントから 取得する：
項目をキューから削除して返します。 オプションのargsブロックがTrue（デフォルト）でtimeoutがNone（デフォルト）の場合、アイテムが利用可能になるまで必要に応じてブロックします。 timeoutが正数の場合、最大timeout秒でブロックし、その時間内に利用可能な項目がなければQueue.Empty例外を発生させます。それ以外の場合（blockはFalse）、すぐに利用可能な場合はその項目を返し、それ以外の場合はQueue.Empty例外を送出します（その場合はタイムアウトは無視されます）。
あなたは通り過ぎているので 0 最初のパラメータとして、それはと同等です get(False)。これはそれを非ブロッキングにします。すぐに値を取得できない場合は、Empty例外が発生し、ワーカープロセスが終了します。すべての "work"関数は同一で、同時にメインキューからプルしようとするため、できない場合があります。すぐに価値を手に入れると死んでしまいます。
を与える .get() 小さなタイムアウトでこの問題は解決するはずです。
2番目の問題は、あなたの "仕事"機能が完了するのに基本的にゼロ時間かかることです。少し休ませて sleep(.2) いくつかの非トライアル作業をシミュレートするために、それはワーカーに分散します。
def my_callable(from_queue, to_queue):
while True:
try:
tmp = from_queue.get(True, .1)
sleep(0.2)
to_queue.put(tmp)
except Empty:
break
編集：
私は言うのを忘れていました、一般的にそれはのタイムアウトに頼らないようにこのタイプの問題のために良いです .get() キューの終わりを知らせるため。 何らかのタイプの「キューの終わり」マーカーオブジェクトを使用してキューに渡すと、作業を終了する時期が来たことをワーカーに知らせることができます。このようにして、それらをすべてブロックして、新しい入力または終了「コマンド」を待つことができます。