Python并发编程线程消息通信机制详解(线程队列)

1Event事件

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()

通过cond来通信，阻塞自己，并使对方执行。从而，达到有顺序的执行。
看下结果

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()

函数会比之前的多一些，同时也从另一方面说明了其功能更加丰富。

我来举个老师点名的例子。

# 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")

运行结果如下