Python3: threading 多线程、锁、线程间通信、定时器、queue 队列

ゝ一纸荒年。 2021-09-21 22:50 360阅读 0赞

本文链接: [https://xiets.blog.csdn.net/article/details/115557891][https_xiets.blog.csdn.net_article_details_115557891]

[Python3 学习笔记（目录）][Python3]

Python 官网 **并发执行** 的文档：[Concurrent Execution][]

# 1. 简单线程实现: \_thread #

Python 官方文档: [\_thread — Low-level threading API][thread _ Low-level threading API]

`_thread` 模块提供了操作多个线程的底层接口，多个控制线程共享全局数据空间。提供了简单的锁机制处理同步问题。

启动线程的函数:

# 启动一个新的线程
    # function: 线程函数, 线程启动后被调用的函数
    # args: 传递给线程函数的参数（或元祖类型的不定长参数）, tuple 类型
    # kwargs: 传递给线程函数的参数（字典类型的不定长参数）, dict 类型
    _thread.start_new_thread(function, args, kwargs=None)

\_thread 启动线程示例:

import time
    import _thread
    
    
    def print_time(tag, *var_args, **var_kwargs):
        count = 5
        while count > 0:
            time.sleep(2)
            count -= 1
            print("%s: %s" % (tag, time.ctime()))
    
    
    # 启动一个新线程, 线程函数为 print_time,
    # 其中 args 元祖中的参数将依次传递给线程函数前面的变量, 剩余的组成新元祖传递给 var_args, 
    # kwargs 字典参数将传递给 var_kwargs, 即结果为:
    # tag: "T1"
    # var_args: (1, "Hello")
    # var_kwargs: {"name": "World", "age": 25}
    _thread.start_new_thread(print_time, ("T1", 1, "Hello"), { "name": "World", "age": 25})
    
    # 再启动一个新线程
    _thread.start_new_thread(print_time, ("T2",))
    
    
    while True:
        pass            # 执行完后需要强制退出

\_thread 的简单锁机制:

# 返回一个新的锁对象
    lock = _thread.allocate_lock()
    
    # 申请获得锁
    lock.acquire(waitflag=1, timeout=-1)
    
    # 释放锁, 锁必须已经被获取过（可以是不同线程获取的）
    lock.release()
    
    # 返回锁的状态, 如果已被某个线程获取则返回 True, 否则返回 False
    lock.locked()

# 2. 高级多线程: threading #

Python 官方文档: [threading — Thread-based parallelism][threading _ Thread-based parallelism]

`threading` 模块在较低级原始的 `_thread` 模块基础上建立了更易用的高级多线程接口。

**threading** 模块中的部分函数:

# 返回当前存活的线程数量
    threading.active_count()
    
    # 以列表形式返回当前所有存活的 线程对象（不包含已结束的线程和尚未启动的线程）
    threading.enumerate()
    
    # 返回当前线程的线程对象
    threading.current_thread()
    
    # 返回主线程的线程对象
    threading.main_thread()
    
    # 返回当前线程的 线程标识符（非零整数）
    threading.get_ident()
    
    # 阻塞函数（ Lock.acquire(), RLock.acquire(), Condition.wait(), ...）中参数 timeout 允许的最大值。
    # 传入超过该值的 timeout 会抛出 OverflowError 异常。
    threading.TIMEOUT_MAX

## 2.1 线程本地数据: [threading.local][] ##

`threading.local` 线程本地数据是特定线程的数据，在不同线程中实例不同。

import threading
    import _thread
    import time
    
    # 创建一个线程本地数据对象, mydata 对象在不同的线程中将会有不同的实例
    mydata = threading.local()
    
    # 在主线程中把 mydata.x 赋值为 1
    mydata.x = 1
    
    
    def thread_func():
        # 在子线程中把 mydata.x 赋值为 2
        mydata.x = 2
    
    
    # 启动一个子线程（在子线程中把 mydata.x 赋值为 2）
    _thread.start_new_thread(thread_func, ())
    
    # 稍作延时, 等待子线程退出
    time.sleep(3)
    
    # 在主线程中获取的 mydata.x 的值将仍然为 1
    print(mydata.x)

## 2.2 线程对象: [threading.Thread][] ##

线程对象 `threading.Thread` 的属性与方法：

# 线程类的构造方法, 创建一个线程对象
    
    # target: 是用于在 run() 方法中被自动调用的可调用对象,
    # name: 线程名称, 默认格式为"Thread-N"
    # daemon: bool 类型, 是否为守护线程, 默认与当前线程相同,
    # True 表示守护线程（没有其他非守护线程时, 进程将退出）
    # args 用于传递给 target 函数的参数元组, 默认是 ()。
    # kwargs 用于传递给 target 函数的参数字典, 默认是 {}。
    #
    class threading.Thread(group=None, target=None, name=None, 
                           args=(), kwargs=None, *, daemon=None)
    
    
    # 启动线程（同一个线程只能启动一次）
    def start()
    
    
    # 线程 start() 启动后运行的方法, 子类中可重写该方法执行线程任务, 
    # 如果 target 不为 None, 该方法中会调用 target()
    def run()
    
    
    # 阻塞等待, 直到线程终结 或 超时指定时间(timeout为浮点秒数)。 
    # 一个线程可以被 join() 多次, join() 当前线程将导致死锁, 
    # join() 操作前线程必须已启动。
    def join(timeout=None)
    
    
    # 获取线程名称
    def getName()
    
    # 设置线程名称
    def setName(name)
    
    
    # 判断线程是否存活, 当 run() 方法刚开始直到 run() 方法刚结束, 该方法返回 True。
    # 即线程启动后结束前表示存活, 即 run() 方法生命周期内。
    def is_alive()
    
    
    # 设置线程是否为守护模式, 必须要再 start() 前设置好。
    # 如果设置为守护线程, 没有其他非守护线程时, 进程将退出。
    def setDaemon(daemonic)
    
    # 是否为守护线程
    def isDaemon()
    
    
    # 线程标识符（非零整数）, 如果线程尚未启动则为 None。与 threading.get_ident() 函数相同。
    # 当线程退出后, 线程标识符可能会被其他新线程复用。即使线程退出后, 仍可得到标识符。
    def ident

多线程实现示例：

import threading
    import time
    
    
    class MyThread(threading.Thread):
        """ 自定义一个线程类, 继承线程基类 """
    
        def run(self):
            """ 启动线程后, run() 方法将在该线程中被调用 """
            time.sleep(1)
            print(self.getName())
    
    
    # 创建两个线程对象
    t1 = MyThread()
    t2 = MyThread()
    
    # 启动线程
    t1.start()
    t2.start()

也可以直接创建 `threading.Thread` 对象并启动，线程启动后将调用 target 参数传递的可调用对象：

import threading
    
    
    def thread_run(*args, **kwargs):
        print("args: ", args)
        print("kwargs: ", kwargs)
        print(threading.current_thread().getName())
    
    
    # 创建线程, 并启动, 启动后 thread_run 将被调用
    threading.Thread(
        target=thread_run,
        args=("Hello", 3.14),
        kwargs={ "name": "World", "age": 20}
    ).start()

## 2.3 线程安全: 锁 ##

锁可被用来保证多线程对共享资源的独占访问，同一时间内只允许一个线程拥有锁，确保线程安全。

### 2.3.1 锁对象: [threading.Lock][] ###

原始锁对象 `threading.Lock` 的属性与方法：

# 原始锁对象 构造方法
    # 实现原始锁对象的类。只要一个线程获得一个锁, 会阻塞随后
    # 尝试获得锁的线程（包括已获得锁的线程）, 直到它被释放; 
    # 任何线程都可以释放它。
    class threading.Lock()
    
    
    # 可以阻塞或非阻塞地获得锁, 
    # blocking=False 表示非阻塞, 默认为阻塞; timeout表示阻塞等待的超时时间, 默认一直等待。
    # 成功获得锁, 则返回 True, 否则返回 False (例如 超时)。
    def acquire(blocking=True, timeout=-1)
    
    
    # 释放一个锁。该方法可以在任何线程中调用, 不单指获得锁的线程。
    def release()
    
    
    # 返回锁的状态, 如果已被某个线程获取则返回 True, 否则返回 False
    def locked()

`threading.Lock` 支持 `with` [上下文管理协议][Link 1]。

### 2.3.2 递归锁对象: [threading.RLock][] ###

`threading.RLock` 重入锁/递归锁对象。

重入锁（RLock）可以被同一个线程多次获取。当一个线程已拥有了重入锁，同一个线程再次获取锁时将不阻塞并直接返回（True）。重入锁必须由获取它的线程释放。

若要锁定锁，线程调用 acquire() 方法；若要解锁，线程调用 release() 方法。线程必须每次获取锁后释放一次（即必须成对使用）。acquire()/release() 对可以嵌套使用，只有最终 release() (最外面一对的 release() ) 将锁解开，才能让其他线程继续处理 acquire() 阻塞。

重入锁(RLock) 与 原始锁(Lock) 拥有相同的方法名称:

class threading.RLock()                     # 重入锁对象 构造方法
    def acquire(blocking=True, timeout=-1)      # 请求获得锁
    def release()                               # 释放锁
    def locked()                                # 返回锁的状态

`threading.RLock` 支持 `with` [上下文管理协议][Link 1]。

### 2.3.3 锁 的使用示例 ###

下面代码需要 10 秒执行完：

import threading
    import time
    
    # 创建一个重入锁对象
    lock = threading.RLock()
    
    
    class MyThread(threading.Thread):
        def run(self):
            lock.acquire()              # 获取锁
            try:
                # SYNC BLOCK # 执行需要同步的相关代码
                time.sleep(5)
            finally:
                lock.release()          # 释放锁
    
    
    # 创建两个线程对象
    t1 = MyThread()
    t2 = MyThread()
    
    # 启动线程
    t1.start()
    t2.start()

锁对象支持 [上下文管理协议][Link 1]，可以为用作 `with` 语句的上下文管理器。当进入 with 语句块时 acquire() 方法会被调用，退出语句块时 release() 会被调用。代码示例：

with lock_obj:
        BLOCK
        # 需要同步的相关代码...

上面代码相当于：

lock_obj.acquire()
    try:
        BLOCK
        # 需要同步的相关代码...
    finally:
        lock_obj.release()

## 2.4 事件对象: [threading.Event][] ##

`threading.Event` 事件对象 用于线程之间的简单通信，一个线程发出事件信号，其他线程等待该信号。

一个事件对象管理一个内部标志，调用 `wait()` 方法进入阻塞直到标志设置为 True（调用时已为 True 将立即返回），调用 `set()` 方法可将标志设置为 True，调用 `clear()` 方法将标志设置为 False。

事件对象 `threading.Event` 的属性和方法：

# 事件对象 的 构造方法
    class threading.Event()
    
    
    # 阻塞线程直到内部标记为 True。当标志为 True 时, 调用该方法将立即返回。
    # 可以设置等待超时时间, 当因超时返回时, 返回值为 False, 否则返回 True。
    def wait(timeout=None)
    
    
    # 将内部标志设置为 True, 所有正在等待这个事件的线程将被唤醒。
    # 当标志为 True 时, 调用 wait() 方法的线程不会被被阻塞。
    def set()
    
    
    # 将内部标志设置为 False, 之后调用 wait() 方法的线程将会被阻塞, 
    # 直到调用 set() 方法将内部标志再次设置为 True。
    def clear()
    
    
    # 当且仅当内部标志为 True 时返回 True
    def is_set()

事件对象 `threading.Event` 代码示例：

import threading
    import time
    
    
    # 创建事件对象
    event = threading.Event()
    
    
    class MyThread01(threading.Thread):
        def run(self):
            event.wait()            # 阻塞, 直到事件标记被设置为 True
            print("MyThread01")
    
    
    class MyThread02(threading.Thread):
        def run(self):
            time.sleep(5)
            event.set()             # 将事件对象标记设置为 True, 唤醒所有在此事件上等待的线程
            print("MyThread02")
    
    
    # 创建线程
    t1 = MyThread01()
    t2 = MyThread02()
    
    # 启动线程
    t1.start()
    t2.start()

## 2.5 条件对象: [threading.Condition][] ##

`threading.Condition` 条件对象 用于线程之间的通信。条件变量总是与某种类型的 **锁对象（Lock/RLock）相关联**，锁对象可以通过参数传入或默认自动创建。条件对象的方法必须要在线程持有关联锁的情况下才能调用。

调用 `wait()` 方法会阻塞线程并释放锁，然后直到超时或被其他线程调用 `notify()` 或 `notify_all()` 方法唤醒，一旦被唤醒，`wait()` 方法将重新请求获取到锁后才能返回。

notify() 和 notify\_all() 方法并不会释放锁，因此被唤醒的线程不会立即从 wait() 方法返回，而是等其他线程释放锁后，调用 wait() 的线程重新竞争获得锁后才能继续。

条件对象 `threading.Condition` 的属性与方法：

# 创建一个 条件对象, 可以指定关联的锁, 默认自动创建。如果传入锁，则锁必须为 Lock 或 RLock 对象。
    # 一个条件对象允许一个或多个线程在被其他线程所通知（notify/notify_all）之前进行等待（wait）。
    class threading.Condition(lock=None)
    
    
    # 请求底层锁
    def acquire(*args)
    
    
    # 释放底层锁
    def release()
    
    
    # 等待直到被通知或发生超时, 该方法会释放底层锁, 然后阻塞, 直到被唤醒或超时。
    # 一旦被唤醒或者超时后, 它将重新获得锁后才能返回, 否则将继续阻塞。
    # 如果线程在调用此方法时没有获得锁, 将会引发 RuntimeError 异常。
    # timeout 为浮点秒数, 默认为始终等待（即不超时）, 发生超时返回 False, 否则返回 True。
    def wait(timeout=None)
    
    
    # 等待（或超时）, 直到条件计算为真。
    #
    # predicate: 一个可调用对象, 并且返回值为 bool 类型
    # 该方法会重复地调用 wait() 直到满足判断式或者发生超时, 
    #
    # 如果忽略超时, 则类似下面代码:
    #
    # while not predicate():
    # condition.wait()
    #
    # timeout: 为浮点秒数, 默认为始终等待（即不超时）。
    #
    # 发生超时返回 False, 否则返回 True。
    #
    def wait_for(predicate, timeout=None)
    
    
    # 最多唤醒 n 个等待这个条件的线程, 默认最多唤醒一个。
    # notify() 不会释放锁, 被唤醒的线程实际上不会返回它调用的 wait(), 直到它可以重新获得锁。
    # 如果调用该方法的线程没有获得底层锁，将抛出 RuntimeError 异常。
    def notify(n=1)
    
    
    # 唤醒所有正在等待这个条件的线程。
    # 该方法行为与 notify() 相似, 但并不只唤醒单一线程，而是唤醒所有等待线程。
    # 如果调用该方法的线程没有获得底层锁，将抛出 RuntimeError 异常。
    def notify_all()

条件对象 `threading.Condition` 代码示例：

import threading
    import time
    
    
    # 创建条件对象, 自动创建底层锁
    cond = threading.Condition()
    
    
    class MyThread01(threading.Thread):
        def run(self):
            cond.acquire()          # 阻塞, 直至获取到底层锁
            try:
                cond.wait()         # 阻塞并释放底层锁, 直至被其他线程 notify 唤醒,
                                    # 唤醒后需要重新竞争获取到底层锁后才能结束阻塞
                print("MyThread01")
            finally:
                cond.release()      # 释放底层锁
    
    
    class MyThread02(threading.Thread):
        def run(self):
            cond.acquire()          # 阻塞, 直至获取到底层锁
            try:
                time.sleep(3)
                cond.notify_all()   # 唤醒所有在此条件对象上等待的线程
                print("MyThread02")
            finally:
                cond.release()      # 释放底层锁
    
    
    # 创建线程
    t1 = MyThread01()
    t2 = MyThread02()
    
    # 启动线程
    t1.start()
    t2.start()

条件对象支持 `with` [上下文管理协议][Link 1]：可以使用 with 语句，在 with 代码块内调用 wait() 或 notify/notify\_all() 方法。acquire() 和 release() 方法会在 with 语句块进入和退出时分别被调用，代码示例：

with condition:
        condition.wait()

上面代码相当于：

condition.acquire()
    try:
        condition.wait()
    finally:
        condition.release()

**`sleep()` 与 `wait()` 方法的区别**：

*  sleep() 与 wait() 调用后都会暂停当前线程并让出 CUP 的执行时间。
 *  time.sleep(secs) 是时间模块的方法，执行不需要占用锁，也不会释放锁，可以在任何地方调用，调用后不需要被唤醒。
 *  threading.Condition.wait(timeout) 是线程模块中的方法，需要获得锁后才能被调用，并且调用后会释放锁，需要被其他线程唤醒或超时并重新获得锁后才能被返回。

## 2.6 信号量对象: [threading.Semaphore][] ##

`threading.Semaphore` 信号量对象。一个信号量内部管理一个计数器，计数器因调用 `acquire()` 方法而递减，因调用 `release()` 方法而递增。计数器的值永远不会小于 0。当调用 acquire() 方法获取信号量时，如果计时器为 0，将会阻塞，直至其他线程调用了 release() 方法使计数器递增。

`threading.Semaphore` 信号量对象的方法：

# 创建一个信号量对象, value 为内部计数器的初始值, 默认为 1,
    # 如果 value < 0 将会抛出 ValueError 异常。
    class threading.Semaphore(value=1)
    
    
    # 获取一个信号量, 获取成功返回 True, 否则返回 False。
    #
    # 如果有信号量(计数器>0), 将立即返回 True, 并将计数器减一。
    # 如果没有信号量则根据参数策略选择是否阻塞。
    #
    # blocking: 是否阻塞, 默认为 blocking=True 阻塞。
    # 如果 blocking=False, 则不阻塞, 如果可以立刻获取到信号量(计数器>0), 则返回 True, 否则返回 False。
    # 
    # timeout: 如果 blocking=False 不阻塞, 则 timeout 必须为 None。
    # 如果 blocking=True, timeout 不为 None, 此时没有信号量(计数器==0), 则最多等待 timeout 秒,
    # 等待结束后如果能获取到信号量, 则返回 True, 并将计数器减一, 否则返回 False。
    #
    # 使用默认参数 blocking=True, timeout=None 表示如果当前没有信号量(计数器==0) 则一直阻塞到获取到信号量才返回。
    #
    acquire(blocking=True, timeout=None)
    
    
    # 释放一个信号量, 将内部计数器的值增加 n, 
    #
    # 如果释放前计数器为 0 且有其他线程在等待获取信号量时, 则唤醒 n 个线程
    #
    release(n=1)

`threading.Semaphore` 允许 release() 调用次数多于 acquire()，即信号量总数可以大于初始化大小。另一个 [threading.BoundedSemaphore][] 有界信号量 则是不允许总信号量超过初始大小。通常情况下，使用信号量对象，一般先 acquire() 再 release() 成对调用（计数器的初始值需 >= 1）。

`threading.Semaphore` 信号量对象 代码示例：

import threading
    import time
    
    
    # 创建信号量对象, value=1 只有一个信号量
    semaphore = threading.Semaphore(value=1)
    
    
    class MyThread01(threading.Thread):
        def run(self):
            # 获取信号量, 获取成功则内部计数器减一。
            # 如果当前没有信号量(内部计数器值为0), 则阻塞到成功获取到信号量。
            semaphore.acquire()
            try:
                time.sleep(1)
                print("MyThread01")
            finally:
                semaphore.release()         # 释放信号量, 内部计数器加一
    
    
    class MyThread02(threading.Thread):
        def run(self):
            # 获取信号量, 获取成功则内部计数器减一。
            # 如果当前没有信号量(内部计数器值为0), 则阻塞到成功获取到信号量。
            semaphore.acquire()
            try:
                time.sleep(1)
                print("MyThread02")
            finally:
                semaphore.release()         # 释放信号量, 内部计数器加一
    
    
    # 创建线程
    t1 = MyThread01()
    t2 = MyThread02()
    
    # 启动线程
    t1.start()
    t2.start()

`threading.Semaphore` 信号量对象支持 `with` [上下文管理协议][Link 1]。代码示例：

with semaphore:
        BLOCK

上面代码相当于：

semaphore.acquire()
    try:
        BLOCK
    finally:
        semaphore.release()

## 2.7 定时器对象: [threading.Timer][] ##

`threading.Timer` 定时器对象，`Timer` 继承自 `Thread`，因此定时器本身是一个线程对象，线程有的方法定时器都有，另外定时器还有取消的方法 cancel()。

# 定时器构造方法
    # interval: 定时器延迟启动时间, start() 后再经过 interval 秒的间隔时间后启动定时器，浮点类型。
    # function: 定时器的任务函数, 定时器执行后将在其线程的 run() 方法中被自动调用。
    # args: 传递给定时器函数的参数（或元祖类型的不定长参数）, tuple 类型。
    # kwargs: 传递给定时器函数的参数（字典类型的不定长参数）, dict 类型。
    class threading.Timer(interval, function, args=None, kwargs=None)
    
    
    # 停止定时器并取消执行计时器将要执行的操作, 仅当计时器仍处于等待状态时有效。
    def cancel()

定时器使用示例：

import threading
    
    
    def timer_func(*args, **kwargs):
        # 输出: timer_func: args=('Hello', 3.14), kwargs={'name': 'World', 'age': 20}
        print(f"timer_func: args={ args}, kwargs={ kwargs}")
    
    
    # 创建定时器, start() 后延迟 3.0 秒执行定时器任务
    t1 = threading.Timer(3.0, timer_func, ("Hello", 3.14), { "name": "World", "age": 20})
    
    # 启动定时器
    t1.start()

# 3. 同步的队列: [queue][] #

`queue` 模块实现了多生产者、多消费者的同步队列。

queue 模块中主要实现了以下 3 种队列：

*  [queue.Queue][]: 先进先出（FIFO）队列
 *  [queue.LifoQueue][]: 后进先出（LIFO）队列
 *  [queue.PriorityQueue][]: 具有优先级顺序的队列

> LifoQueue 和 PriorityQueue 继承自 Queue。

还有一个简单的队列（Python 3.7+）：

*  [queue.SimpleQueue][]: 无界的 FIFO 队列，简单队列，无任务跟踪等高级功能（SimpleQueue 是独立的，不是继承 Queue）。

`queue.Queue`（包括 LifoQueue 和 PriorityQueue）的属性和方法：

# 队列的构造方法,
    # maxsize 表示队列中可放入 项目数 的上限。
    # 当队列中项目数量达到 maxsize 时, put() 操作将阻塞至队列中有项目被 get() 取出。
    # 如果 maxsize<=0, 则队列为无限大。
    class queue.Queue(maxsize=0)
    
    
    # 获取队列的大致大小。
    # qsize() > 0 不保证 get() 不被阻塞。
    # qsize() < maxsize 也不保证 put() 不被阻塞。
    Queue.qsize()
    
    
    # 如果队列为空, 则返回 True, 否则返回 False。
    # 相当于判断 qsize() == 0, 即:
    # empty() 为 False, 不保证 get() 不被阻塞;
    # empty() 为 True, 不保证 put() 不被阻塞。
    Queue.empty()
    
    
    # 如果队列已满, 则返回 True, 否则返回 False。
    # 和 empty(), qsize() 类似, 即:
    # full() 为 True, 不保证 get() 不被阻塞;
    # full() 为 False, 不保证 put() 不被阻塞。
    Queue.full()
    
    
    # 将 item 放入队列, 如果队列已满, 则根据参数决定是否阻塞或抛出异常。
    #
    # block: 是否阻塞, 默认为 block=True 阻塞。
    # 如果 block=False, 则不阻塞, 如果此时队列已满, 将抛出 queue.Full 异常。
    #
    # timeout: 当 block=True 阻塞时才有效, 最多阻塞 timeout 秒, 
    # timeout 秒后队列仍然是满的, 将抛出 queue.Full 异常。
    # timeout=None 表示一直阻塞至队列中有空插槽。
    # timeout<0 将抛出 ValueError 异常。
    #
    Queue.put(item, block=True, timeout=None)
    
    
    # 相当于 put(item, block=False)
    Queue.put_nowait(item)
    
    
    # 从队列中移除并返回一个 item, 如果队列为空, 则根据参数决定是否阻塞或抛出异常。
    #
    # block: 是否阻塞, 默认为 block=True 阻塞。
    # 如果 block=False, 则不阻塞, 如果此时队列为空, 将抛出 queue.Empty 异常。
    #
    # timeout: 当 block=True 阻塞时才有效, 最多阻塞 timeout 秒, 
    # timeout 秒后队列仍然是空的, 将抛出 queue.Empty 异常。
    # timeout=None 表示一直阻塞至队列中有项目。
    # timeout<0 将抛出 ValueError 异常。
    #
    Queue.get(block=True, timeout=None)
    
    
    # 相当于 get(block=False) 
    Queue.get_nowait()
    
    
    # 每个 get() 取出的任务, 后续调用 task_done() 通知队列该任务已处理完。
    Queue.task_done()
    
    
    # 阻塞至队列中的所有任务都被接收并处理完毕。
    # 即所有 put() 进队列的任务（包括阻塞等待的）都被接收, 
    # 并且已 get() 出所有任务并都已完成调用了 task_done(), 此时 join() 阻塞才解除。
    Queue.join()

`LifoQueue` 后进先出队列用法与 `Queue` 相同，`PriorityQueue` 优先级队列的 item 通过以 `(priority_number, item)` 元祖的形式 put 和 get。

## 3.1 [queue.Queue][] 先进先出（FIFO）队列 ##

`queue.Queue` 先进先出（FIFO）队列 代码示例：

import queue
    import threading
    import time
    
    
    class Task:
        def __init__(self, sleep, text):
            self.__sleep = sleep
            self.__text = text
    
        def do_task(self):
            # 执行任务
            time.sleep(self.__sleep)
            print(f"working task ({ self.__text})")
    
    
    # 创建 先进先出（FIFO）队列
    q = queue.Queue()
    
    
    def worker():
        time.sleep(1)
        while True:
            task = q.get()      # 取出任务, 如果当前没任务则阻塞等待
            task.do_task()      # 执行任务
            q.task_done()       # 通知队列任务完成
    
    
    # 创建并启动工作线程
    threading.Thread(target=worker).start()
    
    
    for i in range(5):
        # 添加任务到队列
        task = Task(1.0, f"T{ i}")
        q.put(task)
    
    # 等待所有任务完成
    q.join()
    print("all task done")

输出：

working task (T0)
    working task (T1)
    working task (T2)
    working task (T3)
    working task (T4)
    all task done

## 3.2 [queue.LifoQueue][] 后进先出（LIFO）队列 ##

`queue.LifoQueue` 后进先出（FIFO）队列 代码示例：

import queue
    import threading
    import time
    
    
    class Task:
        def __init__(self, sleep, text):
            self.__sleep = sleep
            self.__text = text
    
        def do_task(self):
            # 执行任务
            time.sleep(self.__sleep)
            print(f"working task ({ self.__text})")
    
    
    # 创建 后进先出（LIFO）队列
    q = queue.LifoQueue()
    
    
    def worker():
        time.sleep(1)
        while True:
            task = q.get()      # 取出任务, 如果当前没任务则阻塞等待
            task.do_task()      # 执行任务
            q.task_done()       # 通知队列任务完成
    
    
    # 创建并启动工作线程
    threading.Thread(target=worker).start()
    
    
    for i in range(5):
        # 添加任务到队列
        task = Task(1.0, f"T{ i}")
        q.put(task)
    
    # 等待所有任务完成
    q.join()
    print("all task done")

输出：

working task (T4)
    working task (T3)
    working task (T2)
    working task (T1)
    working task (T0)
    all task done

## 3.3 [queue.PriorityQueue][] 具有优先级顺序的队列 ##

`queue.PriorityQueue` 优先级队列 代码示例：

import queue
    import threading
    import time
    
    
    class Task:
        def __init__(self, sleep, text):
            self.__sleep = sleep
            self.__text = text
    
        def do_task(self):
            # 执行任务
            time.sleep(self.__sleep)
            print(f"working task ({ self.__text})")
    
    
    # 创建 优先级队列
    q = queue.PriorityQueue()
    
    
    def worker():
        time.sleep(1)
        while True:
            task = q.get()[1]   # 取出任务, 如果当前没任务则阻塞等待
            task.do_task()      # 执行任务
            q.task_done()       # 通知队列任务完成
    
    
    # 创建并启动工作线程
    threading.Thread(target=worker).start()
    
    
    for i in range(5):
        # 创建任务列
        task = Task(1.0, f"T{ i}")
        # 以 (priority_number, task) 元祖的形式添加任务到队列,
        # priority_number 为 int 或 float, 数值越小越优先,
        # 在队列中（未取出）的任务的 priority_number 不能相同
        q.put((i, task))
    
    # 等待所有任务完成
    q.join()
    print("all task done")

输出：

working task (T0)
    working task (T1)
    working task (T2)
    working task (T3)
    working task (T4)
    all task done

## 3.4 [queue.SimpleQueue][] FIFO 简单队列 ##

`queue.SimpleQueue` 是先进先出（FIFO）的，大小无限制的，没有任务跟踪等高级功能的简单队列。

SimpleQueue 类有以下方法（用法与 Queue 相同）：

SimpleQueue.qsize()                                 # 获取队列当前的大小
    SimpleQueue.empty()                                 # 队列是否为空
    
    SimpleQueue.put(item, block=True, timeout=None)     # 添加 item 到队列
    SimpleQueue.put_nowait(item)                        # 不阻塞添加 item 到队列
    
    SimpleQueue.get(block=True, timeout=None)           # 从队列获取 item
    SimpleQueue.get_nowait()                            # 不阻塞从队列获取 item

SimpleQueue 没有 `full()`, `task_done()`, `join()` 方法，因为此队列是无界且无任务跟踪功能的。

SimpleQueue 代码示例：

输出：

working task (T0)
    working task (T1)
    working task (T2)
    working task (T3)
    working task (T4)

[https_xiets.blog.csdn.net_article_details_115557891]: https://xiets.blog.csdn.net/article/details/115557891
[Python3]: https://xiets.blog.csdn.net/article/details/109266600
[Concurrent Execution]: https://docs.python.org/zh-cn/3/library/concurrency.html
[thread _ Low-level threading API]: https://docs.python.org/zh-cn/3/library/_thread.html
[threading _ Thread-based parallelism]: https://docs.python.org/zh-cn/3/library/threading.html
[threading.local]: https://docs.python.org/zh-cn/3/library/threading.html#thread-local-data
[threading.Thread]: https://docs.python.org/zh-cn/3/library/threading.html#thread-objects
[threading.Lock]: https://docs.python.org/zh-cn/3/library/threading.html#lock-objects
[Link 1]: https://docs.python.org/zh-cn/3/library/threading.html#with-locks
[threading.RLock]: https://docs.python.org/zh-cn/3/library/threading.html#rlock-objects
[threading.Event]: https://docs.python.org/zh-cn/3/library/threading.html#event-objects
[threading.Condition]: https://docs.python.org/zh-cn/3/library/threading.html#condition-objects
[threading.Semaphore]: https://docs.python.org/zh-cn/3/library/threading.html#semaphore-objects
[threading.BoundedSemaphore]: https://docs.python.org/zh-cn/3/library/threading.html#threading.BoundedSemaphore
[threading.Timer]: https://docs.python.org/zh-cn/3/library/threading.html#timer-objects
[queue]: https://docs.python.org/zh-cn/3/library/queue.html
[queue.Queue]: https://docs.python.org/zh-cn/3/library/queue.html#queue.Queue
[queue.LifoQueue]: https://docs.python.org/zh-cn/3/library/queue.html#queue.LifoQueue
[queue.PriorityQueue]: https://docs.python.org/zh-cn/3/library/queue.html#queue.PriorityQueue
[queue.SimpleQueue]: https://docs.python.org/zh-cn/3/library/queue.html#queue.SimpleQueue