免费领取：测试资料+测试用例+简历模板+测试文档

本文为霍格沃兹测试开发学社学员学习笔记分享

原文链接：https://ceshiren.com/t/topic/26328

线程基本使用

单线程

def main():    print("在扔一个苹果")
if __name__ == "__main__":    main()

多线程 Python 提供了 thread、threading 等模块来进行线程的创建与管理，后者在线程管理能力上更进一步，因此我们通常使用 threading 模块。创建一个线程需要指定该线程执行的任务（函数名）、以及该函数需要的参数，

import threadingimport timedef apple_1():    print("苹果1")    time.sleep(1)def apple_2():    print("苹果2")    time.sleep(1)def main():    thread = threading.Thread(target=apple_1)    thread2 = threading.Thread(target=apple_2)    thread.start()    thread2.start()    print("苹果3")    print("有多少小小丑？ ", threading.active_count())    print("这些小丑是谁呢？", threading.enumerate())
if __name__ == "__main__":    main()

守护线程

线程是程序执行的最小单位，Python 在进程启动起来后，会自动创建一个主线程，之后使用多线程机制可以在此基础上进行分支，产生新的子线程。子线程启动起来后，主线程默认会等待所有线程执行完成之后再退出。但是我们可以将子线程设置为守护线程，此时主线程任务一旦完成，所有子线程将会和主线程一起结束（就算子线程没有执行完也会退出）。守护线程可以在线程启动之前，通过 setDaemon(True)的形式进行设置，或者在创建子线程对象时，以参数的形式指定：

thread01 = Thread(target=target01, args=“”, name=“线程 1”, daemon=True)

但是需要注意，如果希望主程序不等待任何线程直接退出，只有所有的线程都被设置为守护线程才有用。

设置线程阻塞

我们可以用 join()方法使主线程陷入阻塞，以等待某个线程执行完毕。因此这也是实现线程同步的一种方式。参数 timeout 可以用来设置主线程陷入阻塞的时间，如果线程不是守护线程，即没有设置 daemon 为 True，那么参数 timeout 是无效的，主线程会一直阻塞，直到子线程执行结束。

线程池的使用

在程序运行过程之中，临时创建一个线程需要耗费不小的代价（包括与操作系统的交互部分），尤其是我们只对一个线程分配一个简短的任务，此时，频繁的线程创建将会严重拖垮程序的执行的效率。因此，在这种情形下，我们可以选择采用线程池技术，即通过预先创建几个空闲线程，在需要多线程来处理任务时，将任务分配给一个处于空闲状态的线程，该线程在执行完成后，将会回归空闲状态，而不是直接销毁；而如果申请从线程池中分配一个空闲线程时，遇到所有线程均处于运行状态，则当前线程可以选择阻塞来等待线程资源的空闲。如此一来，程序对于线程的管理将会更加灵活。Python 从 3.2 开始，就将线程池作为内置模块包含了进来，可以通过 concurrent.futures.ThreadPoolExecutor 来调用，使用方法也很简单。

GIL 全局解释器锁

GIL（GlobalInterpreterLock，全局解释器锁）是 CPython 中采用的一种机制，它确保同一时刻只有一个线程在执行 Python 字节码。给整个解释器加锁使得解释器多线程运行更方便，而且开发的 CPython 也更易于维护，但是代价是牺牲了在多处理器上的并行性。因此，在相当多的场景中，CPython 解释器下的多线程机制的性能都不尽如人意

import threadingimport time
def task():    a = 0    while a < 9999*9999:        a += 1

def main():    start_time = time.time()    thread = threading.Thread(target=task)    thread2 = threading.Thread(target=task)    thread2.start()    thread.start()    thread.join()    thread2.join()    task()    print("all time: ", time.time() - start_time)