0Pricing
Python Academy · レッスン

ThreadPoolExecutor

タスクを並行して実行します

「ThreadPoolExecutor」はCoddyKit上の無料Python Academyレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはPython Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 Python Academyコースには全4レッスンが含まれています。

スレッドプールを使う理由

スレッドの作成と結合を手動で行うのは面倒です。concurrent.futures.ThreadPoolExecutorを使うと、ワーカースレッドのプールを管理し、結果を簡単に受け取れます。

from concurrent.futures import ThreadPoolExecutor

def square(n):
    return n * n

with ThreadPoolExecutor() as ex:
    future = ex.submit(square, 5)
    print(future.result())

submitとFuture

submit()は呼び出しをスケジュールし、すぐにFutureを返します。値を取得するには.result()を呼び出します。値の準備ができるまで、この呼び出しはブロックします。

from concurrent.futures import ThreadPoolExecutor

def greet(name):
    return 'hi ' + name

with ThreadPoolExecutor(max_workers=2) as ex:
    f1 = ex.submit(greet, 'Ana')
    f2 = ex.submit(greet, 'Bob')
    print(f1.result())
    print(f2.result())

複数の入力に対するmap

executor.map(func, iterable)は、すべての項目に対して関数を並行して実行し、結果を入力順に生成します。

from concurrent.futures import ThreadPoolExecutor

def cube(n):
    return n ** 3

with ThreadPoolExecutor() as ex:
    results = ex.map(cube, range(5))
    print(list(results))

ワーカー数を制御する

max_workersは、一度に実行できるスレッド数の上限を設定します。I/Oバウンドな処理では、CPUコア数より多くのワーカーを使用できます。

from concurrent.futures import ThreadPoolExecutor
import time

def io_task(n):
    time.sleep(0.05)
    return n

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as ex:
    print(list(ex.map(io_task, range(4))))
print('ran concurrently')

as_completed

as_completed(futures)は、投入された順序に関係なく、各Futureが完了した時点ですぐに生成します。結果を到着次第処理したい場合に便利です。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def work(n):
    time.sleep(0.01 * (3 - n))
    return n

with ThreadPoolExecutor() as ex:
    futures = [ex.submit(work, i) for i in range(3)]
    for f in as_completed(futures):
        print('finished', f.result())

例外を処理する

タスクで例外が発生すると、その例外はFutureに保存され、.result()を呼び出したときに再発生します。その呼び出しをtry/exceptで囲みます。

from concurrent.futures import ThreadPoolExecutor

def risky(n):
    if n == 0:
        raise ValueError('cannot be zero')
    return 10 // n

with ThreadPoolExecutor() as ex:
    f = ex.submit(risky, 0)
    try:
        print(f.result())
    except ValueError as e:
        print('caught:', e)

複数の引数を使ったマッピング

mapは複数のイテラブルを受け取り、組み込み関数のmapと同じように、それらを引数として組み合わせます。

from concurrent.futures import ThreadPoolExecutor

def add(a, b):
    return a + b

with ThreadPoolExecutor() as ex:
    print(list(ex.map(add, [1, 2, 3], [10, 20, 30])))

結果を辞書に収集する

as_completedを使う場合、どの結果がどの入力に対応するか分かるように、各Futureを入力に対応付けるのが一般的です。

from concurrent.futures import ThreadPoolExecutor, as_completed

def length(word):
    return len(word)

words = ['cat', 'tiger', 'ox']
with ThreadPoolExecutor() as ex:
    future_to_word = {ex.submit(length, w): w for w in words}
    out = {}
    for f in as_completed(future_to_word):
        out[future_to_word[f]] = f.result()
print(sorted(out.items()))

コンテキストマネージャーによる終了処理

executorをwithブロックで使用すると、shutdown()が自動的に呼び出され、保留中のすべてのタスクが完了するまで待機します。

from concurrent.futures import ThreadPoolExecutor

def job(n):
    return n * 2

with ThreadPoolExecutor() as ex:
    futures = [ex.submit(job, i) for i in range(3)]
print('all tasks done after the with block')
print([f.result() for f in futures])

Futureの状態を確認する

Futureには.done()と.running()があり、ブロックせずに進行状況を確認できます。

from concurrent.futures import ThreadPoolExecutor

def quick(n):
    return n + 1

with ThreadPoolExecutor() as ex:
    f = ex.submit(quick, 41)
    result = f.result()
    print('done?', f.done())
    print('value:', result)

使用する場面

ThreadPoolExecutorは、HTTP呼び出し、ファイル読み取り、データベースクエリなど、多数の小さなI/Oバウンドタスクに適しています。CPUバウンドな処理には、代わりにProcessPoolExecutorを使用します。

from concurrent.futures import ThreadPoolExecutor

urls = ['a', 'b', 'c']

def fetch(u):
    return 'fetched ' + u

with ThreadPoolExecutor(max_workers=3) as ex:
    for r in ex.map(fetch, urls):
        print(r)

理解度チェック

ThreadPoolExecutorの理解度を確認しましょう。

復習

ThreadPoolExecutorについて学びました。

  • submit()はFutureを返し、map()はイテラブルを順番に処理します。
  • as_completed()は処理が完了した順にfutureを返します。
  • 例外はresult()を通じて発生します。
  • withブロックがシャットダウンを処理します。

次は、multiprocessingによる真の並列処理です。

よくある質問

「ThreadPoolExecutor」レッスンは無料ですか？

はい。「ThreadPoolExecutor」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、Python Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 Python Academyコースには全4レッスンが含まれています。

「ThreadPoolExecutor」で何を学びますか？

タスクを並行して実行します ブラウザで直接実行するハンズオンコードでPython Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

Python Academyを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのPython Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。

「ThreadPoolExecutor」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このPython Academyレッスンでコードを書いて実行できますか？

はい。すべてのPython Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. スレッドとGIL
  2. ThreadPoolExecutor
  3. multiprocessingの基礎
  4. データの安全な共有
← Python Academyに戻る