指令碼開發 / 執行緒池 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 密集型任務通常不會獲得明顯提速。