接口測試-併發處理

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參數，而是在函數內加上超時判斷。

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