接口測試-並發處理

1 多線程執行

import threading
from datetime import *

def test(): print(datetime.now()) def thd(): Theaders = [] for i in range(10): t = threading.Thread(target=test) Theaders.append(t) for t in Theaders: t.start() if __name__ == '__main__': thd()

如果要並發執行N次，建議將並發數拆分成n次，每個線程循環執行n次函數，這樣在啟動下一個線程的時候，上一個線程已經在循環執行了。

import threading
from datetime import *

def test(): print(datetime.now()) def looptest(): for i in range(50): test() def thd(): Threads= [] for i in range(100): t = threading.Thread(target=looptest) Threads.append(t) for t in Threads: t.start() if __name__ == '__main__': thd()

2 守護線程

上面創建的線程是main()線程的子線程，即先啟動主線程main()，然后執行thd自動子線程。

守護線程則是在主線程執行完后，所有的子線程都被關閉（無論子線程是否執行完成）。默認是沒有守護線程，主線程執行完畢之后，會等待子線程全部執行完畢，才會結束程序。如果當子線程死循環，程序將不會關閉，因此在測試時可以設置守護線程解決這個問題。

import threading
from datetime import *

def test(): print(datetime.now()) def looptest(): for i in range(50): test() def thd(): Threads= [] for i in range(1): t = threading.Thread(target=looptest) Threads.append(t) t.setDaemon(True) for t in Threads: t.start() if __name__ == '__main__': thd()

3 阻塞線程

還可以通過子線程join()方法阻塞線程，讓主線程等待子線程完成之后再往下執行，等主線程執行完畢后在關閉所有子線程。

並且可以通過join()的timeout參數來控制，如果超過時間子線程未完成，則主線程繼續執行，執行完成后關閉子線程。

import threading
from datetime import *

def test(): print(datetime.now()) def looptest(): for i in range(50): test() def thd(): Threads= [] for i in range(1): t = threading.Thread(target=looptest) Threads.append(t) t.setDaemon(True) for t in Threads: t.start() for t in Threads: t.join(1) if __name__ == '__main__': thd()

超時機制：完成第一個線程的超時之后開始計算第二個線程的超時，如果執行10個線程超時時間就是10秒。一般不推薦用timeout參數，而是在函數內加上超時判斷。

阻塞線程的主要意義在於控制子線程與主線程的執行順序。