Python线程间通信
目录
线程间通信
在实际场景中，一个团队协作完成共同任务时，成员之间必须通过沟通才能顺利完成工作，这个道理同样适用于线程。在编程中，我们创建多个线程并为每个线程分配不同的子任务，以此减少处理器的空闲时间。因此，线程之间也需要具备通信能力，通过相互交互以同步的方式完成整体任务。
关于线程间通信，需关注以下几个重要点：
- 无性能收益：如果线程和进程之间无法实现有效的通信，并发和并行带来的性能提升将毫无意义。
- 保障任务完成：缺少完善的线程间通信机制，分配的任务将无法被正确完成。
- 比进程间通信更高效：线程间通信比进程间通信更高效、更易用，因为同一进程内的所有线程共享相同的地址空间，无需使用共享内存。
用于线程安全通信的 Python 数据结构
多线程代码中存在一个问题：如何将信息从一个线程传递到另一个线程。标准的通信原语无法解决该问题，因此我们需要实现自定义的复合对象，让线程之间可以安全地共享对象，实现线程安全的通信。以下几种数据结构经过简单改造后，即可支持线程安全的通信：
集合（Set）
要以线程安全的方式使用集合数据结构，我们需要继承集合类，实现自定义的锁机制。
示例
以下 Python 示例演示了如何继承集合类实现线程安全：
class extend_class(set):
def __init__(self, *args, **kwargs):
self._lock = Lock()
super(extend_class, self).__init__(*args, **kwargs)
def add(self, elem):
self._lock.acquire()
try:
super(extend_class, self).add(elem)
finally:
self._lock.release()
def delete(self, elem):
self._lock.acquire()
try:
super(extend_class, self).delete(elem)
finally:
self._lock.release()
在上述示例中，我们定义了一个名为extend_class的类，它继承自 Python 的set类。在该类的构造方法中创建了一个锁对象，同时定义了add()和delete()两个线程安全的方法。这两个方法均基于父类的功能实现，唯一的区别是增加了锁的获取和释放逻辑。
装饰器
使用装饰器是实现线程安全通信的另一种重要方法。
示例
以下 Python 示例演示了装饰器的使用方式：
def lock_decorator(method):
def new_deco_method(self, *args, **kwargs):
with self._lock:
return method(self, *args, **kwargs)
return new_deco_method
class Decorator_class(set):
def __init__(self, *args, **kwargs):
self._lock = Lock()
super(Decorator_class, self).__init__(*args, **kwargs)
@lock_decorator
def add(self, *args, **kwargs):
return super(Decorator_class, self).add(elem)
@lock_decorator
def delete(self, *args, **kwargs):
return super(Decorator_class, self).delete(elem)
在上述示例中，我们定义了一个名为lock_decorator的装饰器方法，随后在Decorator_class的构造方法中创建了锁对象，并为add()和delete()方法添加了该装饰器，让这两个方法实现了线程安全。它们同样基于父类的功能实现，仅增加了装饰器的锁逻辑。
列表（List）
列表是一种线程安全、高效且易用的数据结构，适合作为临时的内存存储容器。在 CPython 中，全局解释器锁（GIL）会保护列表，防止其被并发访问。但需要注意的是，列表本身是线程安全的，但其内部存储的数据却不受保护。
例如，若有另一个线程同时执行L.append(x)操作，该操作无法保证返回预期结果。这是因为尽管append()是原子操作且线程安全，但其他线程对列表数据的并发修改，仍会让输出结果出现竞态条件带来的副作用。
为了解决这类问题并安全地修改数据，我们必须实现完善的锁机制，确保多个线程不会陷入竞态条件。可以像前面的示例一样，通过继承列表类来实现自定义的锁机制。
列表的其他原子操作如下：
L.append(x)
L1.extend(L2)
x = L[i]
x = L.pop()
L1[i:j] = L2
L.sort()
x = y
x.field = y
D[x] = y
D1.update(D2)
D.keys()
说明：
- L、L1、L2 均为列表
- D、D1、D2 均为字典
- x、y 为对象
- i、j 为整数
队列（Queue）
如果列表内的数据不受保护，我们可能会因竞态条件获取或删除错误的数据项，因此推荐使用队列数据结构。
队列的现实例子有很多：单车道的单向公路遵循 “先进先出” 规则，售票处、公交站的排队队伍也是如此。
队列是 Python默认线程安全的数据结构，我们无需手动实现复杂的锁机制。Python 提供了queue模块，让我们可以在应用中使用不同类型的队列。
队列的类型
Python 的queue模块提供了三种队列类型，本节将分别介绍：
- 普通队列（先进先出，FIFO）
- 后进先出队列（LIFO）
- 优先级队列
普通队列（先进先出，FIFO）
这是 Python 最常用的队列实现，遵循 “先进先服务” 的原则，也被直接称为先进先出队列。
Python 实现先进先出队列
Python 中，先进先出队列可基于单线程实现，也可基于多线程实现。
单线程的先进先出队列
实现单线程 FIFO 队列时，可使用queue.Queue类实现基础的先进先出容器：通过put()方法向序列的一端添加元素，通过get()方法从另一端移除元素。
示例
以下 Python 程序实现了单线程的先进先出队列：
import queue
q = queue.Queue()
for i in range(8):
q.put("item-" + str(i))
while not q.empty():
print(q.get(), end = " ")
输出结果
plaintext
item-0 item-1 item-2 item-3 item-4 item-5 item-6 item-7
上述输出表明，单线程环境下，队列中元素的移除顺序与插入顺序完全一致。
多线程的先进先出队列
实现多线程 FIFO 队列时，需先定义基于queue模块的myqueue()函数，其中get()和put()方法的作用与单线程实现中一致；随后声明并实例化多个线程，这些线程会以先进先出的方式消费队列中的元素。
示例
以下 Python 程序实现了多线程的先进先出队列：
import threading
import queue
import random
import time
def myqueue(queue):
while not queue.empty():
item = queue.get()
if item is None:
break
print("{} 从队列中取出 {}".format(threading.current_thread(), item))
queue.task_done()
time.sleep(2)
q = queue.Queue()
for i in range(5):
q.put(i)
threads = []
for i in range(4):
thread = threading.Thread(target=myqueue, args=(q,))
thread.start()
threads.append(thread)
for thread in threads:
thread.join()
输出结果
plaintext
<Thread(Thread-3654, started 5044)> 从队列中取出 0
<Thread(Thread-3655, started 3144)> 从队列中取出 1
<Thread(Thread-3656, started 6996)> 从队列中取出 2
<Thread(Thread-3657, started 2672)> 从队列中取出 3
<Thread(Thread-3654, started 5044)> 从队列中取出 4
后进先出队列（LIFO）
该队列的规则与先进先出队列完全相反，遵循 **“后进先服务”** 的原则，其实现逻辑与栈（Stack）数据结构一致。后进先出队列在实现人工智能中的深度优先搜索等算法时非常实用。
Python 实现后进先出队列
Python 中，后进先出队列同样可基于单线程和多线程实现。
单线程的后进先出队列
实现单线程 LIFO 队列时，使用queue.LifoQueue类实现基础的后进先出容器：通过put()方法将元素添加到容器的头部，通过get()方法也从头部移除元素。
示例
以下 Python 程序实现了单线程的后进先出队列：
import queue
q = queue.LifoQueue()
for i in range(8):
q.put("item-" + str(i))
while not q.empty():
print(q.get(), end=" ")
输出结果
plaintext
item-7 item-6 item-5 item-4 item-3 item-2 item-1 item-0
上述输出表明，单线程环境下，队列中元素的移除顺序与插入顺序完全相反。
多线程的后进先出队列
其实现方式与多线程先进先出队列基本一致，唯一区别是使用queue.LifoQueue类初始化队列，而非queue.Queue。
示例
以下 Python 程序实现了多线程的后进先出队列：
import threading
import queue
import random
import time
def myqueue(queue):
while not queue.empty():
item = queue.get()
if item is None:
break
print("{} 从队列中取出 {}".format(threading.current_thread(), item))
queue.task_done()
time.sleep(2)
q = queue.LifoQueue()
for i in range(5):
q.put(i)
threads = []
for i in range(4):
thread = threading.Thread(target=myqueue, args=(q,))
thread.start()
threads.append(thread)
for thread in threads:
thread.join()
输出结果
plaintext
<Thread(Thread-3882, started 4928)> 从队列中取出 4
<Thread(Thread-3883, started 4364)> 从队列中取出 3
<Thread(Thread-3884, started 6908)> 从队列中取出 2
<Thread(Thread-3885, started 3584)> 从队列中取出 1
<Thread(Thread-3882, started 4928)> 从队列中取出 0
优先级队列
先进先出和后进先出队列中，元素的取出顺序都与插入顺序相关，但实际开发中，很多场景下优先级比插入顺序更重要。
举一个现实例子：机场安检时，不同人群的检查顺序并非按到达时间排序，要客、航空公司工作人员、海关人员等会被优先检查，普通乘客则按排队顺序检查。
优先级队列的另一个重要应用是任务调度器的开发，一种常见的设计思路是：优先处理队列中优先级最高的任务。优先级队列可根据元素的优先级值来获取队列中的元素。
Python 实现优先级队列
Python 中，优先级队列同样可基于单线程和多线程实现。
单线程的优先级队列
实现单线程优先级队列时，使用queue.PriorityQueue类实现基于优先级的任务容器：调用put()方法时为元素绑定一个优先级值，值越小，优先级越高，调用get()方法时会优先取出优先级高的元素。
示例
以下 Python 程序实现了单线程的优先级队列：
import queue as Q
p_queue = Q.PriorityQueue()
p_queue.put((2, '紧急'))
p_queue.put((1, '最紧急'))
p_queue.put((10, '无关紧要'))
p_queue.put((5, '重要'))
while not p_queue.empty():
item = p_queue.get()
print('%s - %s' % item)
输出结果
plaintext
1 - 最紧急
2 - 紧急
5 - 重要
10 - 无关紧要
上述输出表明，队列会根据优先级值存储元素，值越小，优先级越高。
多线程的优先级队列
其实现方式与多线程先进先出、后进先出队列基本一致，主要有两处区别：
- 使用
queue.PriorityQueue类初始化队列； - 队列的生成方式不同，以下示例中会向队列中添加两组完全相同的数据集。
示例
以下 Python 程序实现了多线程的优先级队列：
import threading
import queue
import random
import time
def myqueue(queue):
while not queue.empty():
item = queue.get()
if item is None:
break
print("{} 从队列中取出 {}".format(threading.current_thread(), item))
queue.task_done()
time.sleep(1)
q = queue.PriorityQueue()
for i in range(5):
q.put(i,1)
for i in range(5):
q.put(i,1)
threads = []
for i in range(2):
thread = threading.Thread(target=myqueue, args=(q,))
thread.start()
threads.append(thread)
for thread in threads:
thread.join()
输出结果
plaintext
<Thread(Thread-4939, started 2420)> 从队列中取出 0
<Thread(Thread-4940, started 3284)> 从队列中取出 0
<Thread(Thread-4939, started 2420)> 从队列中取出 1
<Thread(Thread-4940, started 3284)> 从队列中取出 1
<Thread(Thread-4939, started 2420)> 从队列中取出 2
<Thread(Thread-4940, started 3284)> 从队列中取出 2
<Thread(Thread-4939, started 2420)> 从队列中取出 3
<Thread(Thread-4940, started 3284)> 从队列中取出 3
<Thread(Thread-4939, started 2420)> 从队列中取出 4
<Thread(Thread-4940, started 3284)> 从队列中取出 4更多推荐
21
0- 0
已为社区贡献92条内容
所有评论(0)