Python 源码分析：queue 队列模块

Illustrations by Max Dmytriv

作者：weapon，闲来笑浮生悬笔一卷入毫端，朱绂临身可与言者不过二三。
博客：zhihu.com/people/hong-wei-peng
起步
queue 模块提供适用于多线程编程的先进先出（FIFO）数据结构。因为它是线程安全的，所以多个线程很轻松地使用同一个实例。
源码分析
先从初始化的函数来看：

class Queue:
def __init__(self, maxsize=0):
# 设置队列的最大容量
self.maxsize = maxsize
self._init(maxsize)
# 线程锁，互斥变量
self.mutex = threading.Lock()
# 由锁衍生出三个条件变量
self.not_empty = threading.Condition(self.mutex)
self.not_full = threading.Condition(self.mutex)
self.all_tasks_done = threading.Condition(self.mutex)
self.unfinished_tasks = 0
def _init(self, maxsize):
# 初始化底层数据结构
self.queue = deque()

复制代码

从这初始化函数能得到哪些信息呢？首先，队列是可以设置其容量大小的，并且具体的底层存放元素的它使用了

collections.deque()

复制代码

双端列表的数据结构，这使得能很方便的做先进先出操作。这里还特地抽象为

_init

复制代码

函数是为了方便其子类进行覆盖，允许子类使用其他结构来存放元素（比如优先队列使用了 list）。
然后就是线程锁

self.mutex

复制代码

，对于底层数据结构

self.queue

复制代码

的操作都要先获得这把锁；再往下是三个条件变量，这三个 Condition 都以

self.mutex

复制代码

作为参数，也就是说它们共用一把锁；从这可以知道诸如

with self.mutex

复制代码

与

with self.not_empty

复制代码

等都是互斥的。
基于这些锁而做的一些简单的操作：
[code]class Queue:
...
def qsize(self):
      # 返回队列中的元素数
      with self.mutex:
         return self._qsize()

def empty(self):
      # 队列是否为空
      with self.mutex:
         return not self._qsize()

def full(self):
      # 队列是否已满
      with self.mutex:
         return 0 < self.maxsize  0:
            if not block:
                  if self._qsize() >= self.maxsize:
                     raise Full # 如果 block 是 False，并且队列已满，那么抛出 Full 异常
            elif timeout is None:
                  while self._qsize() >= self.maxsize:
                     self.not_full.wait() # 阻塞直到由剩余空间
            elif timeout < 0: # 不合格的参数值，抛出ValueError
                  raise ValueError("'timeout' must be a non-negative number")
            else:
                  endtime = time() + timeout  # 计算等待的结束时间
                  while self._qsize() >= self.maxsize:
                     remaining = endtime - time()
                     if remaining

Python 源码分析：queue 队列模块

浏览过的版块