接口测试-并发处理

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参数，而是在函数内加上超时判断。

阻塞线程的主要意义在于控制子线程与主线程的执行顺序。