Python并发编程：线程间消息通信的三种核心机制
本文深入解析Python中线程间消息通信的三种核心机制：Event、Condition和Queue。通过Event实现简单的触发信号，利用Condition处理复杂的顺序控制，重点讲解Queue在多线程数据安全交换中的高级用法及task_done与join的配合原理，帮助开发者掌握不同场景下的最佳实践。
在前面的内容中，我们已经介绍了如何创建和启动线程。相信读者此时会有一种直观感受：线程操作无非就是实例化对象，然后调用start()方法启动，看似非常基础。
然而，实际工程项目中的并发场景远比示例复杂。线程之间的执行往往存在严格的顺序依赖或条件约束，并非随意启动即可。仅掌握基础的创建与启动知识，不足以应对复杂的并发控制需求。
因此，本文将重点探讨如何精确控制线程的执行时机与顺序。
实现多线程协同控制的本质，在于建立有效的消息通信机制。通过该机制发送指令，明确告知线程何时执行、何时阻塞以及执行具体任务，从而实现有序协作。
经过归纳总结，Python中实现线程通信的主要方法有以下三种：
threading.Event
threading.Condition
queue.Queue
接下来，我们将逐一深入解析这三种机制。
1. Event事件机制
Python提供了Threading.Event这一简洁的通信原语，它基于通用的条件变量概念。其核心逻辑是：多个线程可以等待某一特定事件的发生，一旦该事件被触发，所有等待该事件的线程将被同时激活。
使用Event机制非常直观，主要涉及三个核心方法：
event = threading.Event() # 重置event，使得所有该event事件都处于待命状态 event.clear() # 等待接收event的指令，决定是否阻塞程序执行 event.wait() # 发送event指令，使所有设置该event事件的线程执行 event.set()
通过以下代码示例进行说明。
import time
import threading
class MyThread(threading.Thread):
def __init__(self, name, event):
super().__init__()
self.name = name
self.event = event
def run(self):
print('Thread: {} start at {}'.format(self.name, time.ctime(time.time())))
# 等待event.set()后，才能往下执行
self.event.wait()
print('Thread: {} finish at {}'.format(self.name, time.ctime(time.time())))
threads = []
event = threading.Event()
# 定义五个线程
[threads.append(MyThread(str(i), event)) for i in range(1,5)]
# 重置event，使得event.wait()起到阻塞作用
event.clear()
# 启动所有线程
[t.start() for t in threads]
print('等待5s...')
time.sleep(5)
print('唤醒所有线程...')
event.set()
Thread: 1 start at Sun May 13 20:38:08 2018 Thread: 2 start at Sun May 13 20:38:08 2018 Thread: 3 start at Sun May 13 20:38:08 2018 Thread: 4 start at Sun May 13 20:38:08 2018 等待5s... 唤醒所有线程... Thread: 1 finish at Sun May 13 20:38:13 2018 Thread: 4 finish at Sun May 13 20:38:13 2018 Thread: 2 finish at Sun May 13 20:38:13 2018 Thread: 3 finish at Sun May 13 20:38:13 2018
从运行结果可以看出，所有线程在调用start()启动后，并未立即执行后续逻辑，而是全部阻塞在self.event.wait()处。此时，必须通过调用event.set()发出指令，所有线程才会解除阻塞并继续执行。
2. Condition条件变量
Condition与Event在原理上相似，但功能更为灵活，两者并无本质区别。
掌握Condition同样只需理解几个关键方法。
cond = threading.Condition() # 类似lock.acquire() cond.acquire() # 类似lock.release() cond.release() # 等待指定触发，同时会释放对锁的获取,直到被notify才重新占有琐。 cond.wait() # 发送指定，触发执行 cond.notify()
这里引用一个经典的“捉迷藏”案例来演示其应用。
import threading, time
class Hider(threading.Thread):
def __init__(self, cond, name):
super(Hider, self).__init__()
self.cond = cond
self.name = name
def run(self):
time.sleep(1) #确保先运行Seeker中的方法
self.cond.acquire()
print(self.name + ': 我已经把眼睛蒙上了')
self.cond.notify()
self.cond.wait()
print(self.name + ': 我找到你了哦 ~_~')
self.cond.notify()
self.cond.release()
print(self.name + ': 我赢了')
class Seeker(threading.Thread):
def __init__(self, cond, name):
super(Seeker, self).__init__()
self.cond = cond
self.name = name
def run(self):
self.cond.acquire()
self.cond.wait()
print(self.name + ': 我已经藏好了，你快来找我吧')
self.cond.notify()
self.cond.wait()
self.cond.release()
print(self.name + ': 被你找到了，哎~~~')
cond = threading.Condition()
seeker = Seeker(cond, 'seeker')
hider = Hider(cond, 'hider')
seeker.start()
hider.start()
通过Condition对象进行通信，线程可以主动阻塞自己并唤醒对方，从而实现严格的执行顺序控制。 查看运行结果如下：
hider: 我已经把眼睛蒙上了 seeker: 我已经藏好了，你快来找我吧 hider: 我找到你了 ~_~ hider: 我赢了 seeker: 被你找到了，哎~~~
3. Queue队列机制
最后介绍的是队列，这是本节的核心重点，也是日常开发中使用频率最高的线程通信方式。
在多线程环境中，从一个线程向另一个线程安全发送数据的最佳实践是使用queue库中的队列对象。创建一个被多个线程共享的Queue实例，线程通过调用put()和get()方法向队列中放入和获取元素，从而实现数据交换。
对于Queue类，我们主要关注几个核心方法即可。
from queue import Queue # maxsize默认为0，不受限 # 一旦>0，而消息数又达到限制，q.put()也将阻塞 q = Queue(maxsize=0) # 默认阻塞程序，等待队列消息，可设置超时时间 q.get(block=True, timeout=None) # 发送消息：默认会阻塞程序至队列中有空闲位置放入数据 q.put(item, block=True, timeout=None) # 等待所有的消息都被消费完 q.join() # 通知队列任务处理已经完成，当所有任务都处理完成时，join() 阻塞将会解除 q.task_done()
以下三个辅助方法了解即可，通常在常规开发中较少直接使用：
# 查询当前队列的消息个数 q.qsize() # 队列消息是否都被消费完，返回 True/False q.empty() # 检测队列里消息是否已满 q.full()
相比前两种机制，Queue提供的函数更多，这也侧面反映了其功能更为丰富和强大。
下面通过一个“老师点名”的案例进行演示。
# coding=utf-8
# /usr/bin/env python
'''
Author: wangbm
Email: wongbingming@163.com
Wechat: mrbensonwon
Blog: python-online.cn
公众号：Python编程时光
date: 2020/9/20 下午7:30
desc:
'''
__author__ = 'wangbm'
from queue import Queue
from threading import Thread
import time
class Student:
def __init__(self, name):
self.name = name
def speak(self):
print("{}：到！".format(self.name))
class Teacher:
def __init__(self, queue):
super().__init__()
self.queue=queue
def call(self, student_name):
if student_name == "exit":
print("点名结束，开始上课..")
else:
print("老师：{}来了没？".format(student_name))
# 发送消息，要点谁的名
self.queue.put(student_name)
class CallManager(Thread):
def __init__(self, queue):
super().__init__()
self.students = {}
self.queue = queue
def put(self, student):
self.students.setdefault(student.name, student)
def run(self):
while True:
# 阻塞程序，时刻监听老师，接收消息
student_name = queue.get()
if student_name == "exit":
break
elif student_name in self.students:
self.students[student_name].speak()
else:
print("老师，咱班，没有 {} 这个人".format(student_name))
queue = Queue()
teacher = Teacher(queue=queue)
s1 = Student(name="小明")
s2 = Student(name="小亮")
cm = CallManager(queue)
cm.put(s1)
cm.put(s2)
cm.start()
print('开始点名~')
teacher.call('小明')
time.sleep(1)
teacher.call('小亮')
time.sleep(1)
teacher.call("exit")
程序运行结果如下：
开始点名~ 老师：小明来了没？ 小明：到！ 老师：小亮来了没？ 小亮：到！ 点名结束，开始上课..
此外，Queue还有一个至关重要的方法：Queue.task_done()。
如果不理解其工作原理，在编写程序时极易导致线程死锁或程序卡死。
当我们使用Queue.get()从队列中取出数据后，必须确保该数据被正常消费处理，这一点至关重要。
如果数据未被正常消费，Queue内部会认为该任务仍在执行中。此时若调用Queue.join()，程序将一直阻塞，即使队列中已无其他消息。
如何解决这种因未标记任务完成而导致的持续阻塞问题呢？
解决方法是在正常消费完数据后，显式调用Queue.task_done()，以此告知队列该任务已处理完毕。
当队列内部的任务计数器归零时，调用Queue.join()便不会再阻塞主线程。
若要深入理解这一过程，可参考 http://python.iswbm.com/en/latest/c02/c02_06.html 中关于自定义线程池的实现示例。
4. 总结
学习上述三种通信机制后，我们可以发现Event和Condition是threading模块原生提供的轻量级模块。它们原理简单，功能单一，主要发送True和False这样的布尔型指令，因此仅适用于简单的同步场景。
相比之下，Queue属于更高级的通信模块，能够传输任意类型的数据，包括字符串、字典等复杂对象。其内部实现实际上引用了Condition模块（例如利用put和get函数的阻塞特性），正是基于对Condition的功能扩展，使得Queue具备了更丰富的功能，更能满足实际复杂应用的需求。
以上就是Python并发编程中线程消息通信机制的详细解析，更多关于Python并发线程通信的资料，请关注本站其他相关文章！
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
