スクリプト開発 / スレッドプール DFF.THREAD

DFF.THREAD は、単一の Task 内で IO 集約型関数（例：一括 HTTP リクエスト）を並行実行するために使用します。スレッドプールは DataFlux Func によって管理されます。

API

メソッド / プロパティ	説明
pool_size	明示的に設定されたスレッドプールサイズ。未設定の場合は None
set_pool_size(pool_size)	スレッドプールサイズを設定します。最初の submit(...) の前に呼び出す必要があります
submit(fn, *args, **kwargs)	関数を送信し、結果 Key を返します
get_result(key, wait=True)	指定された結果を取得します
get_all_results(wait=True)	すべての結果を取得します
pop_result(wait=True)	完了済みの結果を 1 つ取り出します。取り出した結果は、他のメソッドでは返されません
is_all_finished	すべて実行完了かどうか
wait_all_finished()	すべての実行が完了するまで待機します

結果オブジェクトの型は DFFThreadResult で、以下の属性を含みます：

属性	説明
key	submit(...) が返す結果 Key
value	関数の戻り値。実行失敗時は通常 None
error	関数がスローした例外。実行成功時は None

set_pool_size(...) は、構成ファイル config.yaml で設定された最大値に照らして検証されます。最初の送信後に再度設定しても有効になりません。明示的に設定しない場合、スレッドプールの作成時には実行時デフォルト値が使用されます。

例

import requests

def fetch(url):
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    return resp.text

@DFF.API('一括リクエスト')
def batch_fetch(urls):
    DFF.THREAD.set_pool_size(10)

    for url in urls:
        DFF.THREAD.submit(fetch, url)

    return [
        {'key': result.key, 'error': repr(result.error)}
        if result.error
        else {'key': result.key, 'value': result.value}
        for result in DFF.THREAD.get_all_results()
    ]

結果の読み取り動作

• get_result(key, wait=False) と pop_result(wait=False) は、一致する準備完了の結果がない場合に None を返します。
• get_all_results(...) は、少なくとも 1 つの関数を送信した後でないと呼び出せません。
• get_all_results(...) は、送信 Key の順に結果を返します。完了順ではありません。
• 完了したものから処理する場合は、pop_result(wait=not DFF.THREAD.is_all_finished) を呼び出すことができます。
• pop_result(...) は互いに独立したタスクに適しています。get_all_results(...) はすべて完了した後にまとめて処理する場合に適しています。

Warning

すべての result.error をチェックする必要があります。そうしないと、スレッドの例外は無視されます。結果がまだ収集されていなくても、Task は送信されたスレッドの処理が停止するまで待ってから終了します。したがって、各スレッド操作には明確なタイムアウトまたはその他の実行境界を設定する必要があります。

スレッドプールは IO 集約型の作業に適しています。CPU 集約型タスクでは、通常、顕著な速度向上は得られません。