本文共 5020 字，大约阅读时间需要 16 分钟。

协程与并发编程：深入理解多线程编程中的关键概念

协程是一种用户态的轻量级线程，由用户程序自己控制调度。与操作系统控制的线程不同，协程在单线程环境下实现并发，能够更高效地利用CPU资源。

协程的特点

信号量（Semaphores）

信号量是一种同步机制，允许多个进程或线程在同一时间竞争有限资源。常用的信号量类型包括信号灯和互斥锁。

信号量代码示例

from threading import Thread, Semaphore
import time
sem = Semaphore(5)  # 最多5个资源
def func():
    sem.acquire()  # 抢锁
    print("Thread %s is working" % Thread.current().name)
    time.sleep(3)  # 模拟资源占用
    sem.release()  # 释放锁
if __name__ == '__main__':
    sem = Semaphore(5)  # 创建5个信号量
    for i in range(21):  # 创建21个线程
        t = Thread(target=func)
        t.start()

例解

• 信号量作用：信号量允许多个进程或线程在同一时间竞争有限资源。例如，创建5个信号量，允许最多5个线程同时访问资源。
• 线程间竞争资源：当信号量被多个线程同时请求时，后续请求会被阻塞，直到当前线程释放信号量。

死锁现象与递归锁

死锁现象

死锁是指一个进程在等待一个资源而被其他进程占用，导致无限等待的状况。死锁不会自行结束，需要程序员主动检测和处理。

递归锁

递归锁允许同一线程多次获得同一锁，通过计数机制实现。其他线程需要等待当前线程释放锁后才能获取锁。

代码示例

from threading import Thread, RLock
import time
mutexA = RLock()
mutexB = RLock()
class MyThread(Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name
    def run(self):
        self.f1()
        self.f2()
    def f1(self):
        mutexA.acquire()
        print("%s acquired A lock" % self.name)
        time.sleep(0.1)
        # 线程1释放B锁后，再次尝试获得B锁
        if not mutexB.is_owned():
            mutexB.acquire()
            print("%s acquired B lock" % self.name)
        else:
            print("线程1 already owns B lock")
        time.sleep(0.1)
        mutexA.release()
    def f2(self):
        if not mutexB.is_owned():
            mutexB.acquire()
            print("%s acquired B lock" % self.name)
            time.sleep(0.1)
            mutexA.acquire()
            print("%s acquired A lock" % self.name)
            mutexA.release()
            mutexB.release()
        else:
            print("线程1 already owns B lock")
if __name__ == '__main__':
    threads = [
        MyThread("Thread 1"),
        MyThread("Thread 2"),
        MyThread("Thread 3"),
        MyThread("Thread 4")
    ]
    for t in threads:
        t.start()

例解

• 递归锁特性：递归锁允许多次递归获取同一锁，线程可以在获取锁后递归调用，锁会自动释放。
• 线程间竞争：线程之间通过递归锁竞争资源，避免死锁发生。

事件（Events）

事件是一种线程间通信机制，用于在一个线程完成任务后通知另一个线程，通常不涉及数据交接。

代码示例

from threading import Event, Thread
import time
import random
event = Event()
def f1():
    print("Thread 1 is running")
    time.sleep(3)
    event.set()
def f2():
    print("Thread 2 is running")
    event.wait()
    print("Thread 2 is done")
if __name__ == '__main__':
    t1 = Thread(target=f1)
    t2 = Thread(target=f2)
    t1.start()
    t2.start()

例解

• 事件机制：事件用于线程间同步，通过设置和等待事件来控制线程的执行。
• 线程间协同：线程可以通过事件进行协同工作，适用于不涉及数据交接的场景。

定时器与线程队列

定时器

定时器是一种在一定时间后自动执行任务的机制，常用于避免阻塞主线程。

代码示例

from threading import Timer
def hello(x):
    print("Hello, World!", x)
timer = Timer(0.2, hello, (10,))  # 0.2秒后执行hello函数
timer.start()

例解

• 定时执行任务：通过设置定时器，线程可以在指定时间后自动执行任务，避免主线程被阻塞。

线程队列

线程队列是一种支持多生产者和消费者的同步结构，支持先进先出、后进先出和优先级队列。

代码示例

import queue
# 先进先出队列
q = queue.Queue(3)
q.put(111)
q.put("aaa")
q.put((1, 2, 3))
print(q.get())  # 111
print(q.get())  # "aaa"
print(q.get())  # (1, 2, 3)
# 后进先出队列
q = queue.LifoQueue(3)
q.put(111)
q.put("aaa")
q.put((1, 2, 3))
print(q.get())  # 111
print(q.get())  # "aaa"
print(q.get())  # (1, 2, 3)
# 优先级队列
q = queue.PriorityQueue(3)
q.put((2, (111,)))
q.put((3, "aaa"))
q.put((1, (1, 2, 3)))
print(q.get())  # (1, (1, 2, 3))
print(q.get())  # (2, (111, ))
print(q.get())  # (3, "aaa")

例解

• 队列类型：线程队列支持多种队列类型，适用于不同的应用场景。
• 数据存取：根据队列类型，数据可以按照先进先出、后进先出或优先级进行存取。

GIL解释器锁

GIL（Global Interpreter Lock）是cpython解释器提供的全局锁，用于保护解释器内存结构。同一进程下只能有一个线程执行python代码，但多个线程可以同时运行本地代码。

例解

• GIL作用：GIL确保同一进程下只能有一个线程执行python代码，防止内存不安全。
• 线程执行：多个线程可以同时执行本地代码，但只能有一个线程执行python代码。

多线程性能测试

多线程在纯计算任务中表现优于多进程，但在纯IO任务中表现一般。正确使用多线程可以提升性能，但需注意避免线程切换带来的开销。

代码示例

from multiprocessing import Process
from threading import Thread
import os, time
def work():
    # res = 0
    # for i in range(100000000):
    #     res *= i
    time.sleep(5)
if __name__ == '__main__':
    start = time.time()
    cpus = os.cpu_count()  # 获取当前CPU核心数
    # 测试多进程
    multi_process_time = 0.0
    for _ in range(cpus):
        p = Process(target=work)
        p.start()
        multi_process_time += 0.1  # 等待子进程启动
    # 测试多线程
    multi_thread_time = 0.0
    for _ in range(cpus):
        t = Thread(target=work)
        t.start()
        multi_thread_time += 0.1
    # 等待所有子进程和线程完成
    for _ in range(cpus):
        p = Process(target=work)
        p.start()
        time.sleep(0.1)
    for _ in range(cpus):
        t = Thread(target=work)
        t.start()
        time.sleep(0.1)
    stop = time.time()
    print(f"Run time with multiple processes: {stop - start - multi_process_time}")
    print(f"Run time with multiple threads: {stop - start - multi_thread_time}")

例解

• 多进程测试：多进程在纯计算任务中表现优于多线程，但需要更多的内存资源。
• 多线程测试：多线程在CPU数量充足时可以更好地利用资源，适合IO密集型任务。

通过上述内容，我们可以全面理解协程、信号量、死锁、递归锁、事件、定时器、线程队列以及多线程性能测试等关键概念，以及如何在实际应用中灵活运用它们。