跳轉到

指令碼開發 / 執行緒池 DFF.THREAD

於 3.3.1 版本新增

DataFlux Func 內建了執行緒池功能 DFF.THREAD，使用者可以直接使用並由系統自動管理執行緒池的開啟和關閉。

對於一些 IO 密集的處理（如大量 HTTP 請求），可以使用執行緒池提高處理效率。

DFF.THREAD.set_pool_size(...)

設定執行緒池大小必須在首次呼叫 DFF.THREAD.submit(...) 之前進行

設定執行緒池大小（執行緒池預設大小為 5）

引數 型別 必須 / 預設值 說明
pool_size int 5 執行緒池大小
示例
1
DFF.THREAD.set_pool_size(10)

DFF.THREAD.submit(...)

使用執行緒池呼叫一個函式

引數 型別 必須 / 預設值 說明
fn function 必須 執行緒執行函式
*args - () 函式位置引數
*kwargs - {} 函式命名引數

返回值為一個字串，用於標記此函式執行的 Key，可在後續配合 DFF.THREAD.get_result(...) 獲取指定任務的結果。

示例
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
import time

def fn(sleep_time):
    time.sleep(sleep_time)
    return sleep_time

def run():
    # 以下兩種呼叫方式等價
    key = DFF.THREAD.submit(fn, 1)
    print(key)

    key = DFF.THREAD.submit(fn, sleep_time=1)
    print(key)

# 輸出：
# thread-result-xxxxx
# thread-result-yyyyy

DFF.THREAD.pop_result(...)

從執行緒池的執行函式結果中彈出一個已完成的結果。

已彈出的函式結果不會再在 DFF.THREAD.get_result(...) 或 DFF.THREAD.get_all_results(...) 中返回

引數 型別 必須 / 預設值 說明
wait bool True 是否等待結果（即是否阻塞）

返回值為 FuncThreadResult 物件。

FuncThreadResult 物件可透過 .value 屬性獲取函式執行返回值，.error 屬性獲取丟擲的錯誤。

示例
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import time

def fn(sleep_time):
    if sleep_time > 2:
        raise Exception('Sleep too long')

    time.sleep(sleep_time)
    return sleep_time

def run():
    DFF.THREAD.submit(fn, 3)
    DFF.THREAD.submit(fn, 2)
    DFF.THREAD.submit(fn, 1)

    while True:
        result = DFF.THREAD.pop_result()
        if result:
            print(f"{result.value}, {repr(result.error)}")
        else:
            break

# 輸出：
# None, Exception('Sleep too long')
# 1, None
# 2, None

DFF.THREAD.get_all_results(...)

獲取全部執行緒執行函式結果。

引數 型別 必須 / 預設值 說明
wait bool True 是否等待結果（即是否阻塞）

返回值為 FuncThreadResult 物件列表。

FuncThreadResult 物件可透過 .value 屬性獲取函式執行返回值，.error 屬性獲取丟擲的錯誤。

示例
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
import time

def fn(sleep_time):
    if sleep_time > 2:
        raise Exception('Sleep too long')

    time.sleep(sleep_time)
    return sleep_time

def run():
    DFF.THREAD.submit(fn, 3)
    DFF.THREAD.submit(fn, 2)
    DFF.THREAD.submit(fn, 1)

    # 獲取全部結果並遍歷
    for result in DFF.THREAD.get_all_results():
        print(f"{result.value}, {repr(result.error)}")

# 輸出：
# None, Exception('Sleep too long')
# 2, None
# 1, None

DFF.THREAD.get_result(...)

獲取某個執行緒執行函式結果

返回值為 FuncThreadResult 物件。

FuncThreadResult 物件可透過 .value 屬性獲取函式執行返回值，.error 屬性獲取丟擲的錯誤。

引數 型別 必須 / 預設值 說明
key str 必須 待獲取結果的函式執行 Key（由 DFF.THREAD.submit(...) 返回）
wait bool True 是否等待結果（即是否阻塞）
示例
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import time

def fn(sleep_time):
    time.sleep(sleep_time)
    return sleep_time

def run():
    key = DFF.THREAD.submit(fn, 1)

    # 獲取指定任務結果
    result = DFF.THREAD.get_result(key=key)
    print(f"{result.value}, {repr(result.error)}")

# 輸出：1, None

DFF.THREAD.is_all_finished

DFF.THREAD.is_all_finished屬性用於判斷當前執行緒池中函式是否全部執行完畢

示例
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import time

def fn(sleep_time):
    time.sleep(sleep_time)
    return sleep_time

def run():
    DFF.THREAD.submit(fn, 3)
    DFF.THREAD.submit(fn, 2)
    DFF.THREAD.submit(fn, 1)

    print(DFF.THREAD.is_all_finished)
    time.sleep(3)
    print(DFF.THREAD.is_all_finished)

# 輸出：
# False
# True

DFF.THREAD.wait_all_finished(...)

判斷當前執行緒池中函式全部執行完畢

示例
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import time

def fn(sleep_time):
    time.sleep(sleep_time)
    return sleep_time

def run():
    DFF.THREAD.submit(fn, 3)
    DFF.THREAD.submit(fn, 2)
    DFF.THREAD.submit(fn, 1)

    DFF.THREAD.wait_all_finished()
    print('Finished')

# 輸出：
# Finished

DFF.THREAD.pop_result(...) VS DFF.THREAD.get_all_results(...)

獲取執行緒池函式執行結果的兩種方式略有不同：

  1. DFF.THREAD.pop_result(...)更適合相互之間獨立的任務，任意一個任務有結果後立即進入後續處理的場景
  2. DFF.THREAD.get_all_results(...)則適合任務之間存在關聯或依賴，需要全部完成後再一起進入後續處理的場景