腳本開發 / 線程池 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)	彈出一個已經完成的結果；彈出後不會再被其他方法返回
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(...)。
• get_all_results(...) 按提交 Key 的順序返回結果，不按完成順序返回。
• 需要邊完成邊處理時，可以調用 pop_result(wait=not DFF.THREAD.is_all_finished)。
• pop_result(...) 適合互相獨立的任務；get_all_results(...) 適合全部完成後統一處理。

Warning

每個 result.error 都必須檢查，否則線程異常會被忽略。即使結果尚未收集，Task 也會等待已提交的線程工作停止後才結束，因此每個線程操作都必須設置明確的超時或其他執行邊界。

線程池適合 IO 密集型工作；CPU 密集型任務通常不會獲得明顯提速。