python的多线程适合计算密集操作_Python多线程操作学习

作者：Kathy-辜 | 来源：互联网 | 2023-07-25 12:58

Python多线程操作什么是线程：线程(Thread)也叫轻量级进程，是操作系统能够进行运算调度的最小单位，它被包涵在进程之中࿰

Python 多线程操作

什么是线程&＃xff1a;

线程(Thread)也叫轻量级进程&＃xff0c;是操作系统能够进行运算调度的最小单位&＃xff0c;它被包涵在进程之中&＃xff0c;是进程中的实际运作单位。线程自己不拥有系统资源&＃xff0c;只拥有一点儿在运行中必不可少的资源&＃xff0c;但它可与同属一个进程的其它线程共享进程所拥有的全部资源。一个线程可以创建和撤消另一个线程&＃xff0c;同一进程中的多个线程之间可以并发执行。

举个简单的例子来理解下&＃xff1a;

假定有一 7 * 24 小时不停工的工厂&＃xff0c;由于其电力有限&＃xff0c;一次仅供一个车间使用&＃xff0c;当一个车间在生产时&＃xff0c;其他车间停工。在这里我们可以理解这个工厂相当于操作系统&＃xff0c;供电设备相当于 CPU&＃xff0c;一个车间相当于一个进程。

一个车间里&＃xff0c;可以有很多工人。他们协同完成一个任务。车间的空间是工人们共享的&＃xff0c;这里一个工人就相当于一个线程&＃xff0c;一个进程可以包括多个线程。比如许多房间是每个工人都可以进出的。这象征一个进程的内存空间是共享的&＃xff0c;每个线程都可以使用这些共享内存。

有时候资源有限&＃xff0c;比如有些房间最多只能容纳一个人&＃xff0c;当一个人占用的时候&＃xff0c;其他人就不能进去&＃xff0c;只能等待。这代表一个线程使用某些共享内存时&＃xff0c;其他线程必须等它结束&＃xff0c;才能使用这一块内存。

一个防止他人进入的简单方法&＃xff0c;就是门口加一把锁。先到的人锁上门&＃xff0c;后到的人看到上锁&＃xff0c;就在门口排队&＃xff0c;等锁打开再进去。这就叫"互斥锁"(Mutual exclusion&＃xff0c;缩写 Mutex )&＃xff0c;防止多个线程同时读写某一块内存区域。

还有些房间&＃xff0c;可以同时容纳 n 个人&＃xff0c;比如厨房。也就是说&＃xff0c;如果人数大于 n&＃xff0c;多出来的人只能在外面等着。这好比某些内存区域&＃xff0c;只能供给固定数目的线程使用。这时的解决方法&＃xff0c;就是在门口挂 n 把钥匙。进去的人就取一把钥匙&＃xff0c;出来时再把钥匙挂回原处。后到的人发现钥匙架空了&＃xff0c;就知道必须在门口排队等着了。这种做法叫做"信号量"( Semaphore )&＃xff0c;用来保证多个线程不会互相冲突。

不难看出&＃xff0c; mutex 是 semaphore 的一种特殊情况(n&＃61;1时)。也就是说&＃xff0c;完全可以用后者替代前者。但是&＃xff0c;因为 mutex 较为简单&＃xff0c;且效率高&＃xff0c;所以在必须保证资源独占的情况下&＃xff0c;还是采用这种设计。

线程有就绪、阻塞、运行三种基本状态。

就绪状态是指线程具备运行的所有条件&＃xff0c;逻辑上可以运行&＃xff0c;在等待处理机&＃xff1b;

运行状态是指线程占有处理机正在运行&＃xff1b;

阻塞状态是指线程在等待一个事件(如某个信号量)&＃xff0c;逻辑上不可执行。

三种状态的相互转化如下图所示&＃xff1a;

image

多线程的优势

那么&＃xff0c;问题来了&＃xff0c;多线程和单线程相比有什么优势呢&＃xff1f;

优势是明显的&＃xff0c;可以提高资源利用率&＃xff0c;让程序响应更快。单线程是按顺序执行&＃xff0c;例如有一单线程程序执行如下操作&＃xff1a;

5秒读取文件A

3秒处理文件A

5秒读取文件B

3秒处理文件B

则需要 16s 完成&＃xff0c;如果开启两个线程来执行&＃xff0c;则如下所示&＃xff1a;

5秒读取文件A

5秒读取文件B &＃43; 3秒处理文件A

3秒处理文件B

则需要 13s 完成。

Python 中的多线程之 GIL

说到 Python 中的多线程&＃xff0c;一个绕不过去的话题就是全局锁 GIL(Global interpreter lock)。GIL 限制了同一时刻只能有一个线程运行&＃xff0c;无法发挥多核 CPU 的优势。首先需要明确的一点是 GIL 并不是 Python 的特性&＃xff0c;它是在实现 Python 解析器(CPython)时所引入的一个概念。就好比 C&＃43;&＃43; 是一套语言(语法)标准&＃xff0c;但是可以用不同的编译器来编译成可执行代码。有名的编译器例如GCC&＃xff0c;INTEL C&＃43;&＃43; &＃xff0c;Visual C&＃43;&＃43;等。Python 也一样&＃xff0c;同样一段代码可以通过 CPython&＃xff0c;PyPy&＃xff0c;Psyco 等不同的 Python 执行环境来执行。像其中的 JPython 就没有GIL。然而因为 CPython 是大部分环境下默认的 Python 执行环境。所以在很多人的概念里 CPython 就是 Python&＃xff0c;也就想当然的把 GIL 归结为 Python 语言的缺陷。所以这里要先明确一点&＃xff1a;GIL 并不是 Python 的特性&＃xff0c;Python 完全可以不依赖于 GIL。

GIL 本质就是一把互斥锁&＃xff0c;既然是互斥锁&＃xff0c;所有互斥锁的本质都一样&＃xff0c;都是将并发运行变成串行&＃xff0c;以此来控制同一时间内共享数据只能被一个任务所修改&＃xff0c;进而保证数据安全。在一个 Python 的进程内&＃xff0c;不仅有主线程或者由该主线程开启的其他线程&＃xff0c;还有解释器开启的垃圾回收等解释器级别的线程&＃xff0c;总之&＃xff0c;所有线程都运行在这一个进程内&＃xff0c;所有数据都是共享的&＃xff0c;这其中&＃xff0c;代码作为一种数据也是被所有线程共享的&＃xff0c;多个线程先访问到解释器的代码&＃xff0c;即拿到执行权限&＃xff0c;然后将 target 的代码交给解释器的代码去执行,

解释器的代码是所有线程共享的&＃xff0c;所以垃圾回收线程也可能访问到解释器的代码而去执行&＃xff0c;这就导致了一个问题:对于同一个数据 100&＃xff0c;可能线程 1 执行 x&＃61;100 的同时&＃xff0c;而垃圾回收执行的是回收 100 的操作&＃xff0c;解决这种问题没有什么高明的方法&＃xff0c;就是加锁处理&＃xff0c;即 GIL。

因此&＃xff0c;有了 GIL 的存在&＃xff0c;同一时刻同一进程中只有一个线程被执行&＃xff0c;那么有人可能要问了&＃xff1a;进程可以利用多核&＃xff0c;而 Python 的多线程却无法利用多核优势&＃xff0c;Python 的多线程是不是没用了&＃xff1f;

答案当然不是。

首先明确我们线程执行的任务是什么&＃xff0c;是做计算(计算密集型)还是做输入输出(I/O 密集型)&＃xff0c;不同地场景使用不同的方法。多核 CPU&＃xff0c;意味着可以有多个核并行完成计算&＃xff0c;所以多核提升的是计算性能,但每个 CPU 一旦遇到 I/O 阻塞&＃xff0c;仍然需要等待&＃xff0c;所以多核对 I/O 密集型任务没什么太高提升。

下面举两个例子来说明&＃xff1a;

例子 1 &＃xff1a;计算密集型任务

计算密集型任务-多进程

from multiprocessing import Process

import os, time

#计算密集型任务

def work():

res &＃61; 0

for i in range(100000000):

res *&＃61; i

if __name__ &＃61;&＃61; "__main__":

l &＃61; []

print("本机为",os.cpu_count(),"核 CPU") # 本机为4核

start &＃61; time.time()

for i in range(4):

p &＃61; Process(target&＃61;work) # 多进程

l.append(p)

p.start()

for p in l:

p.join()

stop &＃61; time.time()

print("计算密集型任务&＃xff0c;多进程耗时 %s" % (stop - start))

运行结果如下

本机为 4 核 CPU

计算密集型任务&＃xff0c;多进程耗时 14.901630640029907

计算密集型任务-多线程

from threading import Thread

import os, time

#计算密集型任务

def work():

res &＃61; 0

for i in range(100000000):

res *&＃61; i

if __name__ &＃61;&＃61; "__main__":

l &＃61; []

print("本机为",os.cpu_count(),"核 CPU") # 本机为4核

start &＃61; time.time()

for i in range(4):

p &＃61; Thread(target&＃61;work) # 多进程

l.append(p)

p.start()

for p in l:

p.join()

stop &＃61; time.time()

print("计算密集型任务&＃xff0c;多线程耗时 %s" % (stop - start))

运行结果如下

本机为 4 核 CPU

计算密集型任务&＃xff0c;多线程耗时 23.559885025024414

例子 2&＃xff1a;I/O 密集型任务

I/O 密集型任务-多进程

from multiprocessing import Process

import os, time

#I/0密集型任务

def work():

time.sleep(2)

print("&＃61;&＃61;&＃61;>", file&＃61;open("tmp.txt", "w"))

if __name__ &＃61;&＃61; "__main__":

l &＃61; []

print("本机为", os.cpu_count(), "核 CPU") # 本机为4核

start &＃61; time.time()

for i in range(400):

p &＃61; Process(target&＃61;work) # 多进程

l.append(p)

p.start()

for p in l:

p.join()

stop &＃61; time.time()

print("I/0密集型任务&＃xff0c;多进程耗时 %s" % (stop - start))

运行结果如下所示&＃xff1a;

本机为 4 核 CPU

I/0密集型任务&＃xff0c;多进程耗时 21.380212783813477

I/O 密集型任务-多线程

from threading import Thread

import os, time

#I/0密集型任务

def work():

time.sleep(2)

print("&＃61;&＃61;&＃61;>", file&＃61;open("tmp.txt", "w"))

if __name__ &＃61;&＃61; "__main__":

l &＃61; []

print("本机为", os.cpu_count(), "核 CPU") # 本机为4核

start &＃61; time.time()

for i in range(400):

p &＃61; Thread(target&＃61;work) # 多线程

l.append(p)

p.start()

for p in l:

p.join()

stop &＃61; time.time()

print("I/0密集型任务&＃xff0c;多线程耗时 %s" % (stop - start))

运行结果如下所示&＃xff1a;

本机为 4 核 CPU

I/0密集型任务&＃xff0c;多线程耗时 2.1127078533172607

结论&＃xff1a;在 Python 中&＃xff0c;对于计算密集型任务&＃xff0c;多进程占优势&＃xff0c;对于 I/O 密集型任务&＃xff0c;多线程占优势。

当然对运行一个程序来说&＃xff0c;随着 CPU 的增多执行效率肯定会有所提高&＃xff0c;这是因为一个程序基本上不会是纯计算或者纯 I/O&＃xff0c;所以我们只能相对的去看一个程序到底是计算密集型还是 I/O 密集型。

Python 多线程之使用方法

Python 提供多线程编程的模块有以下几个&＃xff1a;

_thread

threading

Queue

multiprocessing

下面一一介绍

_thread 模块提供了低级别的基本功能来支持多线程功能&＃xff0c;提供简单的锁来确保同步&＃xff0c;推荐使用 threading 模块。

threading 模块对 _thread 进行了封装&＃xff0c;提供了更高级别&＃xff0c;功能更强&＃xff0c;更易于使用的线程管理的功能&＃xff0c;对线程的支持更为完善&＃xff0c;绝大多数情况下&＃xff0c;只需要使用 threading 这个高级模块就够了。

使用 threading 进行多线程操作:

方法一&＃xff1a;是创建 threading.Thread 实例&＃xff0c;调用其 start() 方法

import time

import threading

def task_thread(counter):

print(f&＃39;线程名称&＃xff1a;{threading.current_thread().name} 参数&＃xff1a;{counter} 开始时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;)

num &＃61; counter

while num:

time.sleep(3)

num -&＃61; 1

print(f&＃39;线程名称&＃xff1a;{threading.current_thread().name} 参数&＃xff1a;{counter} 结束时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;)

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:

print(f&＃39;主线程开始时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;)

#初始化3个线程&＃xff0c;传递不同的参数

t1 &＃61; threading.Thread(target&＃61;task_thread, args&＃61;(3,))

t2 &＃61; threading.Thread(target&＃61;task_thread, args&＃61;(2,))

t3 &＃61; threading.Thread(target&＃61;task_thread, args&＃61;(1,))

#开启三个线程

t1.start()

t2.start()

t3.start()

#等待运行结束

t1.join()

t2.join()

t3.join()

print(f&＃39;主线程结束时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;)

运行结果如下所示

主线程开始时间&＃xff1a;2018-07-06 23:03:46

线程名称&＃xff1a;Thread-1 参数&＃xff1a;3 开始时间&＃xff1a;2018-07-06 23:03:46

线程名称&＃xff1a;Thread-2 参数&＃xff1a;2 开始时间&＃xff1a;2018-07-06 23:03:46

线程名称&＃xff1a;Thread-3 参数&＃xff1a;1 开始时间&＃xff1a;2018-07-06 23:03:46

线程名称&＃xff1a;Thread-3 参数&＃xff1a;1 结束时间&＃xff1a;2018-07-06 23:03:49

线程名称&＃xff1a;Thread-2 参数&＃xff1a;2 结束时间&＃xff1a;2018-07-06 23:03:52

线程名称&＃xff1a;Thread-1 参数&＃xff1a;3 结束时间&＃xff1a;2018-07-06 23:03:55

主线程结束时间&＃xff1a;2018-07-06 23:03:55

方法二&＃xff1a;继承 Thread 类&＃xff0c;在子类中重写 run() 和 init() 方法

import time

import threading

class MyThread(threading.Thread):

def __init__(self, counter):

super().__init__()

self.counter &＃61; counter

def run(self):

print(

f&＃39;线程名称&＃xff1a;{threading.current_thread().name} 参数&＃xff1a;{self.counter} 开始时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;

)

counter &＃61; self.counter

while counter:

time.sleep(3)

counter -&＃61; 1

print(

f&＃39;线程名称&＃xff1a;{threading.current_thread().name} 参数&＃xff1a;{self.counter} 结束时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;

)

if __name__ &＃61;&＃61; "__main__":

print(f&＃39;主线程开始时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;)

# 初始化3个线程&＃xff0c;传递不同的参数

t1 &＃61; MyThread(3)

t2 &＃61; MyThread(2)

t3 &＃61; MyThread(1)

# 开启三个线程

t1.start()

t2.start()

t3.start()

# 等待运行结束

t1.join()

t2.join()

t3.join()

print(f&＃39;主线程结束时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;)

运行结果如下&＃xff0c;与方法一的运行结果一致

主线程开始时间&＃xff1a;2018-07-06 23:34:16

线程名称&＃xff1a;Thread-1 参数&＃xff1a;3 开始时间&＃xff1a;2018-07-06 23:34:16

线程名称&＃xff1a;Thread-2 参数&＃xff1a;2 开始时间&＃xff1a;2018-07-06 23:34:16

线程名称&＃xff1a;Thread-3 参数&＃xff1a;1 开始时间&＃xff1a;2018-07-06 23:34:16

线程名称&＃xff1a;Thread-3 参数&＃xff1a;1 结束时间&＃xff1a;2018-07-06 23:34:19

线程名称&＃xff1a;Thread-2 参数&＃xff1a;2 结束时间&＃xff1a;2018-07-06 23:34:22

线程名称&＃xff1a;Thread-1 参数&＃xff1a;3 结束时间&＃xff1a;2018-07-06 23:34:25

主线程结束时间&＃xff1a;2018-07-06 23:34:25

如果继承 Thread 类&＃xff0c;想调用外部传入函数&＃xff0c;代码如下所示

import time

import threading

def task_thread(counter):

print(f&＃39;线程名称&＃xff1a;{threading.current_thread().name} 参数&＃xff1a;{counter} 开始时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;)

num &＃61; counter

while num:

time.sleep(3)

num -&＃61; 1

print(f&＃39;线程名称&＃xff1a;{threading.current_thread().name} 参数&＃xff1a;{counter} 结束时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;)

class MyThread(threading.Thread):

def __init__(self, target, args):

super().__init__()

self.target &＃61; target

self.args &＃61; args

def run(self):

self.target(*self.args)

if __name__ &＃61;&＃61; "__main__":

print(f&＃39;主线程开始时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;)

# 初始化3个线程&＃xff0c;传递不同的参数

t1 &＃61; MyThread(target&＃61;task_thread,args&＃61;(3,))

t2 &＃61; MyThread(target&＃61;task_thread,args&＃61;(2,))

t3 &＃61; MyThread(target&＃61;task_thread,args&＃61;(1,))

# 开启三个线程

t1.start()

t2.start()

t3.start()

# 等待运行结束

t1.join()

t2.join()

t3.join()

print(f&＃39;主线程结束时间&＃xff1a;{time.strftime("%Y-%m-%d %H:%M:%S")}&＃39;)

这样就和方法一是相通的&＃xff0c;实例化自定义的线程类&＃xff0c;运行结果不变。

线程同步之 Lock (互斥锁)&＃xff1a;

如果多个线程共同对某个数据修改&＃xff0c;则可能出现不可预料的结果&＃xff0c;这个时候就需需要使用互斥锁来进步同步。如下所示的代码&＃xff0c;在三个线程对共同变量 num 进行 100 万次加减操作之后&＃xff0c;其 num 的结果不为 0&＃xff0c;

import time, threading

num &＃61; 0

def task_thread(n):

global num

for i in range(1000000):

num &＃61; num &＃43; n

num &＃61; num - n

t1 &＃61; threading.Thread(target&＃61;task_thread, args&＃61;(6,))

t2 &＃61; threading.Thread(target&＃61;task_thread, args&＃61;(17,))

t3 &＃61; threading.Thread(target&＃61;task_thread, args&＃61;(11,))

t1.start()

t2.start()

t3.start()

t1.join()

t2.join()

t3.join()

print(num)

运行结果如下&＃xff1a;

-19

之所以会出现不为 0 的情况&＃xff0c;因为修改 num 需要多条语句&＃xff0c;当一个线程正在执行 num&＃43;n 时&＃xff0c;另一个线程正在执行 num-m &＃xff0c;从而导致之前的线程执行 num-n 时 num 的值已不是之前的值&＃xff0c;从而导致最终的结果不为 0 。

为了保证数据的正确性&＃xff0c;需要使用互斥锁对多个线程进行同步&＃xff0c;限制当一个线程正在访问数据时&＃xff0c;其他只能等待&＃xff0c;直到前一线程释放锁。使用 threading.Thread 对象的 Lock 和 Rlock 可以实现简单的线程同步&＃xff0c;这两个对象都有 acquire 方法和 release 方法&＃xff0c;对于那些需要每次只允许一个线程操作的数据&＃xff0c;可以将其操作放到 acquire 和 release 方法之间。如下&＃xff1a;

import time, threading

num &＃61; 0

lock &＃61; threading.Lock()

def task_thread(n):

global num

# 获取锁&＃xff0c;用于线程同步

lock.acquire()

for i in range(1000000):

num &＃61; num &＃43; n

num &＃61; num - n

#释放锁&＃xff0c;开启下一个线程

lock.release()

t1 &＃61; threading.Thread(target&＃61;task_thread, args&＃61;(6,))

t2 &＃61; threading.Thread(target&＃61;task_thread, args&＃61;(17,))

t3 &＃61; threading.Thread(target&＃61;task_thread, args&＃61;(11,))

t1.start(); t2.start(); t3.start()

t1.join(); t2.join(); t3.join()

print(num)

运行结果

线程同步之 Semaphore(信号量)

互斥锁同时只允许一个线程访问共享数据&＃xff0c;而信号量是同时允许一定数量的线程访问共享数据&＃xff0c;比如银行柜台有 5 个窗口&＃xff0c;则允许同时有 5 个人办理业务&＃xff0c;后面的人只能等待前面有了办完业务后才可以进入柜台办理。

未例代码如下&＃xff1a;

import threading

import time

# 同时只有5个人办理业务

semaphore &＃61; threading.BoundedSemaphore(5)

# 模拟银行业务办理

def yewubanli(name):

semaphore.acquire()

time.sleep(3)

print(f"{time.strftime(&＃39;%Y-%m-%d %H:%M:%S&＃39;)} {name} 正在办理业务")

semaphore.release()

thread_list &＃61; []

for i in range(12):

t &＃61; threading.Thread(target&＃61;yewubanli, args&＃61;(i,))

thread_list.append(t)

for thread in thread_list:

thread.start()

for thread in thread_list:

thread.join()

# while threading.active_count() !&＃61; 1:

# time.sleep(1)

运行结果如下所示

2018-07-08 12:33:57 4 正在办理业务

2018-07-08 12:33:57 1 正在办理业务

2018-07-08 12:33:57 3 正在办理业务

2018-07-08 12:33:57 0 正在办理业务

2018-07-08 12:33:57 2 正在办理业务

2018-07-08 12:34:00 7 正在办理业务

2018-07-08 12:34:00 5 正在办理业务

2018-07-08 12:34:00 6 正在办理业务

2018-07-08 12:34:00 9 正在办理业务

2018-07-08 12:34:00 8 正在办理业务

2018-07-08 12:34:03 11 正在办理业务

2018-07-08 12:34:03 10 正在办理业务

可以看出&＃xff0c;同一时刻只有 5 个人正在办理业务&＃xff0c;即同一时刻只有5个线程获得资源运行。

线程同步之 Condition

条件对象能让一个线程 A 停下来&＃xff0c;等待其他线程 B &＃xff0c;线程 B 满足了某个条件后通知(notify)线程 A 继续运行。线程首先获取一个条件变量锁&＃xff0c;如果条件不足&＃xff0c;则该线程等待(wait)并释放条件变量锁&＃xff0c;如果满足就执行线程&＃xff0c;也可以通知其他状态为 wait 的线程。其他处于 wait 状态的线程接到通知后会重新判断条件。

下面为一个有趣的例子

import threading

class Boy(threading.Thread):

def __init__(self, cond, name):

super(Boy, self).__init__()

self.cond &＃61; cond

self.name &＃61; name

def run(self):

self.cond.acquire()

print(self.name &＃43; ": 嫁给我吧&＃xff01;&＃xff1f;")

self.cond.notify() # 唤醒一个挂起的线程&＃xff0c;让hanmeimei表态

self.cond.wait() # 释放内部所占用的琐&＃xff0c;同时线程被挂起&＃xff0c;直至接收到通知被唤醒或超时&＃xff0c;等待hanmeimei回答

print(self.name &＃43; ": 我单下跪&＃xff0c;送上戒指&＃xff01;")

self.cond.notify()

self.cond.wait()

print(self.name &＃43; ": Li太太&＃xff0c;你的选择太明治了。")

self.cond.release()

class Girl(threading.Thread):

def __init__(self, cond, name):

super(Girl, self).__init__()

self.cond &＃61; cond

self.name &＃61; name

def run(self):

self.cond.acquire()

self.cond.wait() # 等待Lilei求婚

print(self.name &＃43; ": 没有情调&＃xff0c;不够浪漫&＃xff0c;不答应")

self.cond.notify()

self.cond.wait()

print(self.name &＃43; ": 好吧&＃xff0c;答应你了")

self.cond.notify()

self.cond.release()

cond &＃61; threading.Condition()

boy &＃61; Boy(cond, "LiLei")

girl &＃61; Girl(cond, "HanMeiMei")

girl.start()

boy.start()

运行结果如下&＃xff1a;

LiLei: 嫁给我吧&＃xff01;&＃xff1f;

HanMeiMei: 没有情调&＃xff0c;不够浪漫&＃xff0c;不答应

LiLei: 我单下跪&＃xff0c;送上戒指&＃xff01;

HanMeiMei: 好吧&＃xff0c;答应你了

LiLei: Li太太&＃xff0c;你的选择太明治了。

线程同步之 Event

事件用于线程间通信。一个线程发出一个信号&＃xff0c;其他一个或多个线程等待&＃xff0c;调用 event 对象的 wait 方法&＃xff0c;线程则会阻塞等待&＃xff0c;直到别的线程 set 之后&＃xff0c;才会被唤醒。上面求婚哥的例子使用 Event 代码如下&＃xff1a;

import threading, time

class Boy(threading.Thread):

def __init__(self, cond, name):

super(Boy, self).__init__()

self.cond &＃61; cond

self.name &＃61; name

def run(self):

print(self.name &＃43; ": 嫁给我吧&＃xff01;&＃xff1f;")

self.cond.set() # 唤醒一个挂起的线程&＃xff0c;让hanmeimei表态

time.sleep(0.5)

self.cond.wait()

print(self.name &＃43; ": 我单下跪&＃xff0c;送上戒指&＃xff01;")

self.cond.set()

time.sleep(0.5)

self.cond.wait()

self.cond.clear()

print(self.name &＃43; ": Li太太&＃xff0c;你的选择太明治了。")

class Girl(threading.Thread):

def __init__(self, cond, name):

super(Girl, self).__init__()

self.cond &＃61; cond

self.name &＃61; name

def run(self):

self.cond.wait() # 等待Lilei求婚

self.cond.clear()

print(self.name &＃43; ": 没有情调&＃xff0c;不够浪漫&＃xff0c;不答应")

self.cond.set()

time.sleep(0.5)

self.cond.wait()

print(self.name &＃43; ": 好吧&＃xff0c;答应你了")

self.cond.set()

cond &＃61; threading.Event()

boy &＃61; Boy(cond, "LiLei")

girl &＃61; Girl(cond, "HanMeiMei")

boy.start()

girl.start()

运行结果如下&＃xff1a;

LiLei: 嫁给我吧&＃xff01;&＃xff1f;

HanMeiMei: 没有情调&＃xff0c;不够浪漫&＃xff0c;不答应

HanMeiMei: 好吧&＃xff0c;答应你了

LiLei: 我单下跪&＃xff0c;送上戒指&＃xff01;

LiLei: Li太太&＃xff0c;你的选择太明治了

线程优先级队列(queue)

Python 的 queue 模块中提供了同步的、线程安全的队列类&＃xff0c;包括先进先出队列 Queue&＃xff0c;后进先出队列 LifoQueue&＃xff0c;和优先级队列 PriorityQueue。这些队列都实现了锁原语&＃xff0c;可以直接使用来实现线程间的同步。

举一个简单的例子&＃xff0c;假如有一小冰箱用来存放冷饮&＃xff0c;假如该小冰箱只能放 5 个冷饮&＃xff0c;A 不停地往冰箱放冷饮&＃xff0c;B 不停地从冰箱取冷饮&＃xff0c;A 和 B 的放取速度可能不一致&＃xff0c;如何保持他们的同步呢&＃xff1f; 这里队列就派上了用场。

先看代码

import threading,time

import queue

#先进先出

q &＃61; queue.Queue(maxsize&＃61;5)

#q &＃61; queue.LifoQueue(maxsize&＃61;3)

#q &＃61; queue.PriorityQueue(maxsize&＃61;3)

def ProducerA():

count &＃61; 1

while True:

q.put(f"冷饮 {count}")

print(f"A 放入:[冷饮 {count}]")

count &＃43;&＃61;1

time.sleep(1)

def ConsumerB():

while True:

print(f"B 取出 [{q.get()}]")

time.sleep(5)

p &＃61; threading.Thread(target&＃61;ProducerA)

c &＃61; threading.Thread(target&＃61;ConsumerB)

c.start()

p.start()

运行结果如下&＃xff1a;

16:29:19 A 放入:[冷饮 1]

16:29:19 B 取出 [冷饮 1]

16:29:20 A 放入:[冷饮 2]

16:29:21 A 放入:[冷饮 3]

16:29:22 A 放入:[冷饮 4]

16:29:23 A 放入:[冷饮 5]

16:29:24 B 取出 [冷饮 2]

16:29:24 A 放入:[冷饮 6]

16:29:25 A 放入:[冷饮 7]

16:29:29 B 取出 [冷饮 3]

16:29:29 A 放入:[冷饮 8]

16:29:34 B 取出 [冷饮 4]

16:29:34 A 放入:[冷饮 9]

以上代码是实现生产者和消费者模型的一个最简单的例子。在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。如果生产者处理速度很快&＃xff0c;而消费者处理速度很慢&＃xff0c;那么生产者就必须等待消费者处理完&＃xff0c;才能继续生产数据。同样的道理&＃xff0c;如果消费者的处理能力大于生产者&＃xff0c;那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。生产者消费者模式是通过一个容器(队列)来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯&＃xff0c;而通过阻塞队列来进行通讯&＃xff0c;所以生产者生产完数据之后不用等待消费者处理&＃xff0c;直接扔给阻塞队列&＃xff0c;消费者不找生产者要数据&＃xff0c;而是直接从阻塞队列里取&＃xff0c;阻塞队列就相当于一个缓冲区&＃xff0c;平衡了生产者和消费者的处理能力。

image

multiprocessing

Python中线程与进程使用的同一模块 multiprocessing。使用方法也基本相同&＃xff0c;唯一不同的是&＃xff0c;from multiprocessing import Pool 这样导入的 Pool 表示的是进程池&＃xff0c;from multiprocessing.dummy import Pool这样导入的 Pool表示的是线程池。这样就可以实现线程里面的并发了。

线程池实例&＃xff1a;

from multiprocessing.dummy import Pool as ThreadPool

import time

def fun(n):

time.sleep(2)

start &＃61; time.time()

for i in range(5):

fun(i)

print("单线程顺序执行耗时:", time.time() - start)

start2 &＃61; time.time()

# 开8个 worker&＃xff0c;没有参数时默认是 cpu 的核心数

pool &＃61; ThreadPool(processes&＃61;2)