python开两个守护线程_每周一个Python模块|threading

作者：哈哈哈阿笑 | 来源：互联网 | 2023-08-27 14:34

其实在Python中，多线程是不推荐使用的，除非明确不支持使用多进程的场景，否则的话，能用多进程就用多进程吧。写这篇文章的目

其实在 Python 中&＃xff0c;多线程是不推荐使用的&＃xff0c;除非明确不支持使用多进程的场景&＃xff0c;否则的话&＃xff0c;能用多进程就用多进程吧。写这篇文章的目的&＃xff0c;可以对比多进程的文章来看&＃xff0c;有很多相通的地方&＃xff0c;看完也许会对并发编程有更好的理解。

GIL

Python(特指 CPython)的多线程的代码并不能利用多核的优势&＃xff0c;而是通过著名的全局解释锁(GIL)来进行处理的。如果是一个计算型的任务&＃xff0c;使用多线程 GIL 就会让多线程变慢。我们举个计算斐波那契数列的例子&＃xff1a;

# coding&＃61;utf-8import timeimport threadingdef profile(func): def wrapper(*args, **kwargs): import time start &＃61; time.time() func(*args, **kwargs) end &＃61; time.time() print &＃39;COST: {}&＃39;.format(end - start) return wrapperdef fib(n): if n<&＃61; 2: return 1 return fib(n-1) &＃43; fib(n-2)&＃64;profiledef nothread(): fib(35) fib(35)&＃64;profiledef hasthread(): for i in range(2): t &＃61; threading.Thread(target&＃61;fib, args&＃61;(35,)) t.start() main_thread &＃61; threading.currentThread() for t in threading.enumerate(): if t is main_thread: continue t.join()nothread()hasthread()# output# COST: 5.05716490746# COST: 6.75599503517

运行的结果你猜猜会怎么样&＃xff1f;还不如不用多线程&＃xff01;

GIL 是必须的&＃xff0c;这是 Python 设计的问题&＃xff1a;Python 解释器是非线程安全的。这意味着当从线程内尝试安全的访问Python 对象的时候将有一个全局的强制锁。在任何时候&＃xff0c;仅仅一个单一的线程能够获取 Python 对象或者 C API。每 100 个字节的 Python 指令解释器将重新获取锁&＃xff0c;这(潜在的)阻塞了 I/O 操作。因为锁&＃xff0c;CPU 密集型的代码使用线程库时&＃xff0c;不会获得性能的提高(但是当它使用之后介绍的多进程库时&＃xff0c;性能可以获得提高)。

那是不是由于 GIL 的存在&＃xff0c;多线程库就是个「鸡肋」呢&＃xff1f;当然不是。事实上我们平时会接触非常多的和网络通信或者数据输入/输出相关的程序&＃xff0c;比如网络爬虫、文本处理等等。这时候由于网络情况和 I/O 的性能的限制&＃xff0c;Python 解释器会等待读写数据的函数调用返回&＃xff0c;这个时候就可以利用多线程库提高并发效率了。

线程对象

先说一个非常简单的方法&＃xff0c;直接使用 Thread 来实例化目标函数&＃xff0c;然后调用 start() 来执行。

import threadingdef worker(): """thread worker function""" print(&＃39;Worker&＃39;)threads &＃61; []for i in range(5): t &＃61; threading.Thread(target&＃61;worker) threads.append(t) t.start() # output# Worker# Worker# Worker# Worker# Worker

生成线程时可以传递参数给线程&＃xff0c;什么类型的参数都可以。下面这个例子只传了一个数字&＃xff1a;

import threadingdef worker(num): """thread worker function""" print(&＃39;Worker: %s&＃39; % num)threads &＃61; []for i in range(5): t &＃61; threading.Thread(target&＃61;worker, args&＃61;(i,)) threads.append(t) t.start() # output# Worker: 0# Worker: 1# Worker: 2# Worker: 3# Worker: 4

还有一种创建线程的方法&＃xff0c;通过继承 Thread 类&＃xff0c;然后重写 run() 方法&＃xff0c;代码如下&＃xff1a;

import threadingimport loggingclass MyThread(threading.Thread): def run(self): logging.debug(&＃39;running&＃39;)logging.basicConfig( level&＃61;logging.DEBUG, format&＃61;&＃39;(%(threadName)-10s) %(message)s&＃39;,)for i in range(5): t &＃61; MyThread() t.start() # output# (Thread-1 ) running# (Thread-2 ) running# (Thread-3 ) running# (Thread-4 ) running# (Thread-5 ) running

因为传递给 Thread 构造函数的参数 args 和 kwargs 被保存成了带 __ 前缀的私有变量&＃xff0c;所以在子线程中访问不到&＃xff0c;所以在自定义线程类中&＃xff0c;要重新构造函数。

import threadingimport loggingclass MyThreadWithArgs(threading.Thread): def __init__(self, group&＃61;None, target&＃61;None, name&＃61;None, args&＃61;(), kwargs&＃61;None, *, daemon&＃61;None): super().__init__(group&＃61;group, target&＃61;target, name&＃61;name, daemon&＃61;daemon) self.args &＃61; args self.kwargs &＃61; kwargs def run(self): logging.debug(&＃39;running with %s and %s&＃39;, self.args, self.kwargs)logging.basicConfig( level&＃61;logging.DEBUG, format&＃61;&＃39;(%(threadName)-10s) %(message)s&＃39;,)for i in range(5): t &＃61; MyThreadWithArgs(args&＃61;(i,), kwargs&＃61;{&＃39;a&＃39;: &＃39;A&＃39;, &＃39;b&＃39;: &＃39;B&＃39;}) t.start() # output# (Thread-1 ) running with (0,) and {&＃39;b&＃39;: &＃39;B&＃39;, &＃39;a&＃39;: &＃39;A&＃39;}# (Thread-2 ) running with (1,) and {&＃39;b&＃39;: &＃39;B&＃39;, &＃39;a&＃39;: &＃39;A&＃39;}# (Thread-3 ) running with (2,) and {&＃39;b&＃39;: &＃39;B&＃39;, &＃39;a&＃39;: &＃39;A&＃39;}# (Thread-4 ) running with (3,) and {&＃39;b&＃39;: &＃39;B&＃39;, &＃39;a&＃39;: &＃39;A&＃39;}# (Thread-5 ) running with (4,) and {&＃39;b&＃39;: &＃39;B&＃39;, &＃39;a&＃39;: &＃39;A&＃39;}

确定当前线程

每个 Thread 都有一个名称&＃xff0c;可以使用默认值&＃xff0c;也可以在创建线程时指定。

import threadingimport timedef worker(): print(threading.current_thread().getName(), &＃39;Starting&＃39;) time.sleep(0.2) print(threading.current_thread().getName(), &＃39;Exiting&＃39;)def my_service(): print(threading.current_thread().getName(), &＃39;Starting&＃39;) time.sleep(0.3) print(threading.current_thread().getName(), &＃39;Exiting&＃39;)t &＃61; threading.Thread(name&＃61;&＃39;my_service&＃39;, target&＃61;my_service)w &＃61; threading.Thread(name&＃61;&＃39;worker&＃39;, target&＃61;worker)w2 &＃61; threading.Thread(target&＃61;worker) # use default namew.start()w2.start()t.start()# output# worker Starting# Thread-1 Starting# my_service Starting# worker Exiting# Thread-1 Exiting# my_service Exiting

守护线程

默认情况下&＃xff0c;在所有子线程退出之前&＃xff0c;主程序不会退出。有些时候&＃xff0c;启动后台线程运行而不阻止主程序退出是有用的&＃xff0c;例如为监视工具生成“心跳”的任务。

要将线程标记为守护程序&＃xff0c;在创建时传递 daemon&＃61;True 或调用set_daemon(True)&＃xff0c;默认情况下&＃xff0c;线程不是守护进程。

import threadingimport timeimport loggingdef daemon(): logging.debug(&＃39;Starting&＃39;) time.sleep(0.2) logging.debug(&＃39;Exiting&＃39;)def non_daemon(): logging.debug(&＃39;Starting&＃39;) logging.debug(&＃39;Exiting&＃39;)logging.basicConfig( level&＃61;logging.DEBUG, format&＃61;&＃39;(%(threadName)-10s) %(message)s&＃39;,)d &＃61; threading.Thread(name&＃61;&＃39;daemon&＃39;, target&＃61;daemon, daemon&＃61;True)t &＃61; threading.Thread(name&＃61;&＃39;non-daemon&＃39;, target&＃61;non_daemon)d.start()t.start()# output# (daemon ) Starting# (non-daemon) Starting# (non-daemon) Exiting

输出不包含守护线程的 Exiting&＃xff0c;因为在守护线程从 sleep() 唤醒之前&＃xff0c;其他线程&＃xff0c;包括主程序都已经退出了。

如果想等守护线程完成工作&＃xff0c;可以使用 join() 方法。

import threadingimport timeimport loggingdef daemon(): logging.debug(&＃39;Starting&＃39;) time.sleep(0.2) logging.debug(&＃39;Exiting&＃39;)def non_daemon(): logging.debug(&＃39;Starting&＃39;) logging.debug(&＃39;Exiting&＃39;)logging.basicConfig( level&＃61;logging.DEBUG, format&＃61;&＃39;(%(threadName)-10s) %(message)s&＃39;,)d &＃61; threading.Thread(name&＃61;&＃39;daemon&＃39;, target&＃61;daemon, daemon&＃61;True)t &＃61; threading.Thread(name&＃61;&＃39;non-daemon&＃39;, target&＃61;non_daemon)d.start()t.start()d.join()t.join()# output# (daemon ) Starting# (non-daemon) Starting# (non-daemon) Exiting# (daemon ) Exiting

输出信息已经包括守护线程的 Exiting。

默认情况下&＃xff0c;join()无限期地阻止。也可以传一个浮点值&＃xff0c;表示等待线程变为非活动状态的秒数。如果线程未在超时期限内完成&＃xff0c;则join()无论如何都会返回。

import threadingimport timeimport loggingdef daemon(): logging.debug(&＃39;Starting&＃39;) time.sleep(0.2) logging.debug(&＃39;Exiting&＃39;)def non_daemon(): logging.debug(&＃39;Starting&＃39;) logging.debug(&＃39;Exiting&＃39;)logging.basicConfig( level&＃61;logging.DEBUG, format&＃61;&＃39;(%(threadName)-10s) %(message)s&＃39;,)d &＃61; threading.Thread(name&＃61;&＃39;daemon&＃39;, target&＃61;daemon, daemon&＃61;True)t &＃61; threading.Thread(name&＃61;&＃39;non-daemon&＃39;, target&＃61;non_daemon)d.start()t.start()d.join(0.1)print(&＃39;d.isAlive()&＃39;, d.isAlive())t.join()# output# (daemon ) Starting# (non-daemon) Starting# (non-daemon) Exiting# d.isAlive() True

由于传递的超时小于守护程序线程休眠的时间&＃xff0c;因此join() 返回后线程仍处于“活动”状态。

枚举所有线程

enumerate() 方法可以返回活动 Thread 实例列表。由于该列表包括当前线程&＃xff0c;并且由于加入当前线程会引入死锁情况&＃xff0c;因此必须跳过它。

import randomimport threadingimport timeimport loggingdef worker(): """thread worker function""" pause &＃61; random.randint(1, 5) / 10 logging.debug(&＃39;sleeping %0.2f&＃39;, pause) time.sleep(pause) logging.debug(&＃39;ending&＃39;)logging.basicConfig( level&＃61;logging.DEBUG, format&＃61;&＃39;(%(threadName)-10s) %(message)s&＃39;,)for i in range(3): t &＃61; threading.Thread(target&＃61;worker, daemon&＃61;True) t.start()main_thread &＃61; threading.main_thread()for t in threading.enumerate(): if t is main_thread: continue logging.debug(&＃39;joining %s&＃39;, t.getName()) t.join() # output# (Thread-1 ) sleeping 0.20# (Thread-2 ) sleeping 0.30# (Thread-3 ) sleeping 0.40# (MainThread) joining Thread-1# (Thread-1 ) ending# (MainThread) joining Thread-3# (Thread-2 ) ending# (Thread-3 ) ending# (MainThread) joining Thread-2

计时器线程

Timer() 在延迟时间后开始工作&＃xff0c;并且可以在该延迟时间段内的任何时间点取消。

import threadingimport timeimport loggingdef delayed(): logging.debug(&＃39;worker running&＃39;)logging.basicConfig( level&＃61;logging.DEBUG, format&＃61;&＃39;(%(threadName)-10s) %(message)s&＃39;,)t1 &＃61; threading.Timer(0.3, delayed)t1.setName(&＃39;t1&＃39;)t2 &＃61; threading.Timer(0.3, delayed)t2.setName(&＃39;t2&＃39;)logging.debug(&＃39;starting timers&＃39;)t1.start()t2.start()logging.debug(&＃39;waiting before canceling %s&＃39;, t2.getName())time.sleep(0.2)logging.debug(&＃39;canceling %s&＃39;, t2.getName())t2.cancel()logging.debug(&＃39;done&＃39;)# output# (MainThread) starting timers# (MainThread) waiting before canceling t2# (MainThread) canceling t2# (MainThread) done# (t1 ) worker running

此示例中的第二个计时器不会运行&＃xff0c;并且第一个计时器似乎在主程序完成后运行的。由于它不是守护线程&＃xff0c;因此在完成主线程时会隐式调用它。

同步机制

Semaphore

在多线程编程中&＃xff0c;为了防止不同的线程同时对一个公用的资源(比如全部变量)进行修改&＃xff0c;需要进行同时访问的数量(通常是 1)。信号量同步基于内部计数器&＃xff0c;每调用一次 acquire()&＃xff0c;计数器减 1&＃xff1b;每调用一次 release()&＃xff0c;计数器加 1。当计数器为 0 时&＃xff0c;acquire() 调用被阻塞。

import loggingimport randomimport threadingimport timeclass ActivePool: def __init__(self): super(ActivePool, self).__init__() self.active &＃61; [] self.lock &＃61; threading.Lock() def makeActive(self, name): with self.lock: self.active.append(name) logging.debug(&＃39;Running: %s&＃39;, self.active) def makeInactive(self, name): with self.lock: self.active.remove(name) logging.debug(&＃39;Running: %s&＃39;, self.active)def worker(s, pool): logging.debug(&＃39;Waiting to join the pool&＃39;) with s: name &＃61; threading.current_thread().getName() pool.makeActive(name) time.sleep(0.1) pool.makeInactive(name)logging.basicConfig( level&＃61;logging.DEBUG, format&＃61;&＃39;%(asctime)s (%(threadName)-2s) %(message)s&＃39;,)pool &＃61; ActivePool()s &＃61; threading.Semaphore(2)for i in range(4): t &＃61; threading.Thread( target&＃61;worker, name&＃61;str(i), args&＃61;(s, pool), ) t.start() # output# 2016-07-10 10:45:29,398 (0 ) Waiting to join the pool# 2016-07-10 10:45:29,398 (0 ) Running: [&＃39;0&＃39;]# 2016-07-10 10:45:29,399 (1 ) Waiting to join the pool# 2016-07-10 10:45:29,399 (1 ) Running: [&＃39;0&＃39;, &＃39;1&＃39;]# 2016-07-10 10:45:29,399 (2 ) Waiting to join the pool# 2016-07-10 10:45:29,399 (3 ) Waiting to join the pool# 2016-07-10 10:45:29,501 (1 ) Running: [&＃39;0&＃39;]# 2016-07-10 10:45:29,501 (0 ) Running: []# 2016-07-10 10:45:29,502 (3 ) Running: [&＃39;3&＃39;]# 2016-07-10 10:45:29,502 (2 ) Running: [&＃39;3&＃39;, &＃39;2&＃39;]# 2016-07-10 10:45:29,607 (3 ) Running: [&＃39;2&＃39;]# 2016-07-10 10:45:29,608 (2 ) Running: []

在这个例子中&＃xff0c;ActivePool() 类只是为了展示在同一时刻&＃xff0c;最多只有两个线程在运行。

Lock

Lock 也可以叫做互斥锁&＃xff0c;其实相当于信号量为 1。我们先看一个不加锁的例子&＃xff1a;

import timefrom threading import Threadvalue &＃61; 0def getlock(): global value new &＃61; value &＃43; 1 time.sleep(0.001) # 使用sleep让线程有机会切换 value &＃61; newthreads &＃61; []for i in range(100): t &＃61; Thread(target&＃61;getlock) t.start() threads.append(t)for t in threads: t.join()print value# 16

不加锁的情况下&＃xff0c;结果会远远的小于 100。那我们加上互斥锁看看&＃xff1a;

import timefrom threading import Thread, Lockvalue &＃61; 0lock &＃61; Lock()def getlock(): global value with lock: new &＃61; value &＃43; 1 time.sleep(0.001) value &＃61; newthreads &＃61; []for i in range(100): t &＃61; Thread(target&＃61;getlock) t.start() threads.append(t)for t in threads: t.join()print value# 100

RLock

acquire() 能够不被阻塞的被同一个线程调用多次。但是要注意的是 release() 需要调用与 acquire() 相同的次数才能释放锁。

先看一下使用 Lock 的情况&＃xff1a;

import threadinglock &＃61; threading.Lock()print(&＃39;First try :&＃39;, lock.acquire())print(&＃39;Second try:&＃39;, lock.acquire(0))# output# First try : True# Second try: False

在这种情况下&＃xff0c;第二次调用将 acquire() 被赋予零超时以防止它被阻塞&＃xff0c;因为第一次调用已获得锁定。

再看看用RLock替代的情况。

import threadinglock &＃61; threading.RLock()print(&＃39;First try :&＃39;, lock.acquire())print(&＃39;Second try:&＃39;, lock.acquire(0))# output# First try : True# Second try: True

Condition

一个线程等待特定条件&＃xff0c;而另一个线程发出特定条件满足的信号。最好说明的例子就是「生产者/消费者」模型&＃xff1a;

import timeimport threadingdef consumer(cond): t &＃61; threading.currentThread() with cond: cond.wait() # wait()方法创建了一个名为waiter的锁&＃xff0c;并且设置锁的状态为locked。这个waiter锁用于线程间的通讯 print &＃39;{}: Resource is available to consumer&＃39;.format(t.name)def producer(cond): t &＃61; threading.currentThread() with cond: print &＃39;{}: Making resource available&＃39;.format(t.name) cond.notifyAll() # 释放waiter锁&＃xff0c;唤醒消费者condition &＃61; threading.Condition()c1 &＃61; threading.Thread(name&＃61;&＃39;c1&＃39;, target&＃61;consumer, args&＃61;(condition,))c2 &＃61; threading.Thread(name&＃61;&＃39;c2&＃39;, target&＃61;consumer, args&＃61;(condition,))p &＃61; threading.Thread(name&＃61;&＃39;p&＃39;, target&＃61;producer, args&＃61;(condition,))c1.start()time.sleep(1)c2.start()time.sleep(1)p.start()# output# p: Making resource available# c2: Resource is available to consumer# c1: Resource is available to consumer

可以看到生产者发送通知之后&＃xff0c;消费者都收到了。

Event

一个线程发送/传递事件&＃xff0c;另外的线程等待事件的触发。我们同样的用「生产者/消费者」模型的例子&＃xff1a;

# coding&＃61;utf-8import timeimport threadingfrom random import randintTIMEOUT &＃61; 2def consumer(event, l): t &＃61; threading.currentThread() while 1: event_is_set &＃61; event.wait(TIMEOUT) if event_is_set: try: integer &＃61; l.pop() print &＃39;{} popped from list by {}&＃39;.format(integer, t.name) event.clear() # 重置事件状态 except IndexError: # 为了让刚启动时容错 passdef producer(event, l): t &＃61; threading.currentThread() while 1: integer &＃61; randint(10, 100) l.append(integer) print &＃39;{} appended to list by {}&＃39;.format(integer, t.name) event.set() # 设置事件 time.sleep(1)event &＃61; threading.Event()l &＃61; []threads &＃61; []for name in (&＃39;consumer1&＃39;, &＃39;consumer2&＃39;): t &＃61; threading.Thread(name&＃61;name, target&＃61;consumer, args&＃61;(event, l)) t.start() threads.append(t)p &＃61; threading.Thread(name&＃61;&＃39;producer1&＃39;, target&＃61;producer, args&＃61;(event, l))p.start()threads.append(p)for t in threads: t.join() # output# 77 appended to list by producer1# 77 popped from list by consumer1# 46 appended to list by producer1# 46 popped from list by consumer2# 43 appended to list by producer1# 43 popped from list by consumer2# 37 appended to list by producer1# 37 popped from list by consumer2# 33 appended to list by producer1# 33 popped from list by consumer2# 57 appended to list by producer1# 57 popped from list by consumer1

可以看到事件被 2 个消费者比较平均的接收并处理了。如果使用了 wait() 方法&＃xff0c;线程就会等待我们设置事件&＃xff0c;这也有助于保证任务的完成。

Queue

队列在并发开发中最常用的。我们借助「生产者/消费者」模式来理解&＃xff1a;生产者把生产的「消息」放入队列&＃xff0c;消费者从这个队列中对去对应的消息执行。

大家主要关心如下 4 个方法就好了&＃xff1a;

put: 向队列中添加一个项。
get: 从队列中删除并返回一个项。
task_done: 当某一项任务完成时调用。
join: 阻塞直到所有的项目都被处理完。

# coding&＃61;utf-8import timeimport threadingfrom random import randomfrom Queue import Queueq &＃61; Queue()def double(n): return n * 2def producer(): while 1: wt &＃61; random() time.sleep(wt) q.put((double, wt))def consumer(): while 1: task, arg &＃61; q.get() print arg, task(arg) q.task_done()for target in(producer, consumer): t &＃61; threading.Thread(target&＃61;target) t.start()

这就是最简化的队列架构。

Queue 模块还自带了 PriorityQueue(带有优先级)和 LifoQueue(后进先出)2 种特殊队列。我们这里展示下线程安全的优先级队列的用法&＃xff0c;PriorityQueue 要求我们 put 的数据的格式是(priority_number, data)&＃xff0c;我们看看下面的例子&＃xff1a;

import timeimport threadingfrom random import randintfrom Queue import PriorityQueueq &＃61; PriorityQueue()def double(n): return n * 2def producer(): count &＃61; 0 while 1: if count > 5: break pri &＃61; randint(0, 100) print &＃39;put :{}&＃39;.format(pri) q.put((pri, double, pri)) # (priority, func, args) count &＃43;&＃61; 1def consumer(): while 1: if q.empty(): break pri, task, arg &＃61; q.get() print &＃39;[PRI:{}] {} * 2 &＃61; {}&＃39;.format(pri, arg, task(arg)) q.task_done() time.sleep(0.1)t &＃61; threading.Thread(target&＃61;producer)t.start()time.sleep(1)t &＃61; threading.Thread(target&＃61;consumer)t.start()# output# put :84# put :86# put :16# put :93# put :14# put :93# [PRI:14] 14 * 2 &＃61; 28# # [PRI:16] 16 * 2 &＃61; 32# [PRI:84] 84 * 2 &＃61; 168# [PRI:86] 86 * 2 &＃61; 172# [PRI:93] 93 * 2 &＃61; 186# [PRI:93] 93 * 2 &＃61; 186

其中消费者是故意让它执行的比生产者慢很多&＃xff0c;为了节省篇幅&＃xff0c;只随机产生 5 次随机结果。可以看到 put 时的数字是随机的&＃xff0c;但是 get 的时候先从优先级更高(数字小表示优先级高)开始获取的。

线程池

面向对象开发中&＃xff0c;大家知道创建和销毁对象是很费时间的&＃xff0c;因为创建一个对象要获取内存资源或者其它更多资源。无节制的创建和销毁线程是一种极大的浪费。那我们可不可以把执行完任务的线程不销毁而重复利用呢&＃xff1f;仿佛就是把这些线程放进一个池子&＃xff0c;一方面我们可以控制同时工作的线程数量&＃xff0c;一方面也避免了创建和销毁产生的开销。

线程池在标准库中其实是有体现的&＃xff0c;只是在官方文章中基本没有被提及&＃xff1a;

In : from multiprocessing.pool import ThreadPoolIn : pool &＃61; ThreadPool(5)In : pool.map(lambda x: x**2, range(5))Out: [0, 1, 4, 9, 16]

当然我们也可以自己实现一个&＃xff1a;

# coding&＃61;utf-8import timeimport threadingfrom random import randomfrom Queue import Queuedef double(n): return n * 2class Worker(threading.Thread): def __init__(self, queue): super(Worker, self).__init__() self._q &＃61; queue self.daemon &＃61; True self.start() def run(self): while 1: f, args, kwargs &＃61; self._q.get() try: print &＃39;USE: {}&＃39;.format(self.name) # 线程名字 print f(*args, **kwargs) except Exception as e: print e self._q.task_done()class ThreadPool(object): def __init__(self, num_t&＃61;5): self._q &＃61; Queue(num_t) # Create Worker Thread for _ in range(num_t): Worker(self._q) def add_task(self, f, *args, **kwargs): self._q.put((f, args, kwargs)) def wait_complete(self): self._q.join()pool &＃61; ThreadPool()for _ in range(8): wt &＃61; random() pool.add_task(double, wt) time.sleep(wt)pool.wait_complete()# output# USE: Thread-1# 1.58762376489# USE: Thread-2# 0.0652918738849# USE: Thread-3# 0.997407997138# USE: Thread-4# 1.69333900685# USE: Thread-5# 0.726900613676# USE: Thread-1# 1.69110052253# USE: Thread-2# 1.89039743989# USE: Thread-3# 0.96281118122

线程池会保证同时提供 5 个线程工作&＃xff0c;但是我们有 8 个待完成的任务&＃xff0c;可以看到线程按顺序被循环利用了。