Pythonの `concurrent.futures`：完了順に先物を反復する

Question

executor.mapと似たものが欲しいですが、結果を反復するときを除いて、私はそれらを順番に反復したいと思います。最初に完了した作業項目が反復などの最初に表示されます。これは、シーケンス内のすべての単一作業項目がまだ終了していない場合は反復処理がブロックされるためです。

私は自分自身のキューを使用してこれを実装する方法を知っているが、私はそれがfuturesフレームワークを使用して可能ですかと思いまして。

（I主に使用するスレッドベースの執行、私はこれらに当てはまる答えが、同様に歓迎されるであろう一般的な答えをしたいと思います。）

UPDATE：答えてくれてありがとう！ as_completedとexecutor.mapをどうすればいいのか説明してください。 executor.mapは、先物を使うときに私にとって最も便利で簡潔なツールです。私はFutureのオブジェクトを手作業で使い始めるのを嫌っています。

Answer 1

executor.map()は、組み込みmap()のように、唯一の反復可能な順序で結果を返すので、残念ながらあなたは完成の順序を決定するためにそれを使用することはできません。 concurrent.futures.as_completed()あなたが探しているものである - ここでは例です：もちろん

import time 
import concurrent.futures 

times = [3, 1, 2] 

def sleeper(secs): 
    time.sleep(secs) 
    print('I slept for {} seconds'.format(secs)) 
    return secs 

# returns in the order given 
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: 
    print(list(executor.map(sleeper, times))) 

# I slept for 1 seconds 
# I slept for 2 seconds 
# I slept for 3 seconds 
# [3, 1, 2] 

# returns in the order completed 
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: 
    futs = [executor.submit(sleeper, secs) for secs in times] 
    print([fut.result() for fut in concurrent.futures.as_completed(futs)]) 

# I slept for 1 seconds 
# I slept for 2 seconds 
# I slept for 3 seconds 
# [1, 2, 3] 

あなたはマップ・インターフェースを使用する必要がある場合、あなたは（おそらくサブクラスに追加上記をカプセル化し、独自のmap_as_completed()機能を作成することができますExecutor()）、私は先物のインスタンスをexecutor.submit()で作成する方がシンプルでクリーンな方法であると考えています（また、引数なし、kwargsも提供できます）。

Answer 2

同時先物は完了の時間に基づいて、イテレータを返します - これは、それはあなたが探していたまさにだように聞こえます。

http://docs.python.org/dev/library/concurrent.futures.html#concurrent.futures.as_completed

あなたが任意の混乱や困難WRTの実装を持っているなら、私に知らせてください。敬具、-David

Answer 3

From python doc

concurrent.futures.as_completed(fs, timeout=None)¶ 

は、彼らが完了すると先物を生み出すFSによって与えられた未来のインスタンスの反復子 が（おそらく別のエグゼキュータ インスタンスが作成した）（ を終え返します。またはキャンセルされた）。 as_completed（） が呼び出される前に完了した先物が最初に返されます。 次（）が呼び出された場合に返されるイテレータは TimeoutErrorを上昇させ、その結果がas_completedに元の通話から、タイムアウト秒後 は使用できません（）。 timeoutはintまたはfloatにすることができます。 timeoutが指定されていないかNoneである場合は、 待ち時間に制限はありません。

あなたはexecutor.map()とexecutor.submit()の違いを理解する必要があります。最初の関数は、関数を引数のベクトルにマップします。それはmapとかなり似ていますが、非同期にタスクを起動します。 submit(func, arg)は、各コールで1つのタスクを起動します。このタスクではfuncがargに適用されます。ここで

は私のpython 3上で実行することができsubmit()とas_completed()を使用するための一例です。0

from concurrent import futures 
import urllib.request 

URLS = ['http://www.foxnews.com/', 
     'http://www.cnn.com/', 
     'http://europe.wsj.com/', 
     'http://www.bbc.co.uk/', 
     'http://some-made-up-domain.com/'] 

def load_url(url, timeout): 
    return urllib.request.urlopen(url, timeout=timeout).read() 

def main(): 
    with futures.ThreadPoolExecutor(max_workers=5) as executor: 
     future_to_url = dict(
      (executor.submit(load_url, url, 60), url) 
      for url in URLS) 

     for future in futures.as_completed(future_to_url): 
      url = future_to_url[future] 
      try: 
       print('%r page is %d bytes' % (
          url, len(future.result()))) 
      except Exception as e: 
       print('%r generated an exception: %s' % (
          url, e)) 

if __name__ == '__main__': 
    main() 

なしmap()は、タスクがsubmitで実行され、as_completed()

は、彼らの完全なよう 先物を生み出すFSによって与えられた未来のインスタンスの反復子を返します（完成またはキャンセルされた）、ここで使用されています。