深入理解Python多线程编程 threading

深入理解Python多线程编程 threading

flyfish

Python 中 threading 模块的概念

threading 模块是 Python 标准库中的一个模块，用于创建和管理线程。线程是轻量级的进程，允许程序在同一进程中并行执行多个任务。每个线程共享相同的内存空间，因此它们可以访问同一进程中的所有资源。

作用

1. 提高响应性：在I/O密集型任务中（如网络请求、文件读写），使用多线程可以让程序在等待I/O操作完成时继续执行其他任务，从而提高程序的响应速度。
2. 简化并发编程：通过线程可以简化并发编程模型，使得编写同时处理多个任务的程序变得更加容易。
3. 资源共享：所有线程共享同一个进程的内存空间，因此它们可以方便地共享数据，但这也需要小心处理同步问题以避免竞态条件。

单线程示例：打印 1 到 10

虽然单线程并不是 threading 模块的主要用途，但我们可以通过它来理解基本的线程概念。

import threading
import time

def print_numbers():
    """打印 1 到 10 的函数"""
    for i in range(1, 11):
        print(f"数字: {i}")
        time.sleep(0.5)  # 模拟耗时操作

# 创建一个线程对象
thread = threading.Thread(target=print_numbers)

# 启动线程
thread.start()

# 等待线程完成
thread.join()

print("主线程结束")

代码解释

• threading.Thread(target=print_numbers)：创建一个新的线程对象，并指定该线程的目标函数为 print_numbers。
• thread.start()：启动线程，开始执行目标函数 print_numbers。
• thread.join()：阻塞主线程，直到新创建的线程执行完毕。
• time.sleep(0.5)：模拟耗时操作，使程序暂停0.5秒，以便我们可以看到输出的延迟效果。

多线程示例：打印偶数和奇数

下面是一个使用 threading 模块创建两个线程的示例，其中一个线程打印0到10之间的偶数，另一个线程打印1到9之间的奇数。

import threading
import time

def print_even_numbers():
    """打印 0 到 10 之间的偶数"""
    for i in range(0, 11, 2):
        print(f"偶数: {i}")
        time.sleep(0.5)

def print_odd_numbers():
    """打印 1 到 9 之间的奇数"""
    for i in range(1, 10, 2):
        print(f"奇数: {i}")
        time.sleep(0.5)

if __name__ == "__main__":
    # 创建两个线程
    even_thread = threading.Thread(target=print_even_numbers)
    odd_thread = threading.Thread(target=print_odd_numbers)

    # 启动两个线程
    even_thread.start()
    odd_thread.start()

    # 等待两个线程完成
    even_thread.join()
    odd_thread.join()

    print("主线程结束")

代码解释

• even_thread = threading.Thread(target=print_even_numbers) 和 odd_thread = threading.Thread(target=print_odd_numbers)：分别创建两个线程对象，一个用于打印偶数，另一个用于打印奇数。
• even_thread.start() 和 odd_thread.start()：启动两个线程，开始执行各自的目标函数。
• even_thread.join() 和 odd_thread.join()：阻塞主线程，直到两个子线程都执行完毕。

执行过程

1. 线程启动：当程序运行时，首先创建了两个线程对象 even_thread 和 odd_thread，然后分别调用 start() 方法启动这两个线程。
2. 并发执行：两个线程几乎同时开始执行各自的 print_even_numbers 和 print_odd_numbers 函数。
3. 交替输出：由于每个线程内部都有 time.sleep(0.5)，这会导致每次循环后线程暂停0.5秒，从而使两个线程有机会交替输出偶数和奇数。
4. 线程结束：当两个线程都完成了各自的循环后，主线程会继续执行 print("主线程结束")。

线程同步的重要性

在多线程编程中，多个线程可能会同时访问和修改共享资源（如变量、数据结构等）。如果没有适当的同步机制，可能会导致竞态条件（Race Condition）或数据不一致的问题。例如：

• 竞态条件：两个或多个线程同时读取和修改同一个变量，导致最终结果不可预测。
• 数据不一致：一个线程在读取或写入数据时被另一个线程中断，导致数据不完整或不一致。

因此，线程同步是确保多个线程安全地访问共享资源的关键。它通过控制线程的执行顺序来避免这些问题。

常见的同步机制

线程同步是多线程编程中的关键部分，确保多个线程安全地访问共享资源。常见的同步机制包括：

• 锁（Lock）：确保同一时刻只有一个线程可以访问共享资源。
• 信号量（Semaphore）：控制对共享资源的最大并发访问数量。
• 条件变量（Condition）：允许线程在某些条件满足时进行等待，并在条件改变时唤醒等待的线程。
• 事件（Event）：一种简单的线程间通信机制，允许一个线程等待某个特定事件的发生。

1. 锁（Lock）

锁是最基本的同步机制之一。锁可以确保同一时刻只有一个线程能够访问共享资源。

示例：使用 threading.Lock 实现线程同步

import threading

# 创建一个全局计数器
counter = 0
lock = threading.Lock()

def increment_counter():
    global counter
    for _ in range(100000):
        with lock:
            # 获取锁后才允许修改计数器
            local_counter = counter
            local_counter += 1
            counter = local_counter
            print(f"当前计数器值: {counter}")

# 创建两个线程
t1 = threading.Thread(target=increment_counter)
t2 = threading.Thread(target=increment_counter)

# 启动线程
t1.start()
t2.start()

# 等待线程完成
t1.join()
t2.join()

print(f"最终计数器值: {counter}")

解释：

• lock = threading.Lock()：创建一个锁对象。
• with lock:：使用上下文管理器自动获取和释放锁。当一个线程进入 with lock: 块时，它会尝试获取锁；如果成功，则继续执行；否则，等待直到锁可用。
• counter：共享资源，两个线程都试图对其进行修改。通过锁机制，确保每次只有一个线程可以修改 counter，从而避免了竞态条件。

2. 信号量（Semaphore）

信号量是一种更高级的同步机制，允许指定数量的线程同时访问某个资源。它可以用于限制对共享资源的并发访问次数。

示例：使用 threading.Semaphore 控制并发访问

import threading
import time

# 创建一个信号量，最多允许3个线程同时访问
semaphore = threading.Semaphore(3)

def access_resource(thread_id):
    print(f"线程 {thread_id} 正在等待信号量...")
    with semaphore:
        print(f"线程 {thread_id} 获得了信号量")
        time.sleep(2)  # 模拟耗时操作
        print(f"线程 {thread_id} 释放了信号量")

# 创建多个线程
threads = [threading.Thread(target=access_resource, args=(i,)) for i in range(5)]

# 启动所有线程
for t in threads:
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

print("所有线程完成")

解释：

• semaphore = threading.Semaphore(3)：创建一个信号量对象，最多允许3个线程同时访问共享资源。
• with semaphore:：使用上下文管理器自动获取和释放信号量。当一个线程进入 with semaphore: 块时，它会尝试获取信号量；如果成功，则继续执行；否则，等待直到信号量可用。
• time.sleep(2)：模拟耗时操作，使得其他线程有机会竞争信号量。

3. 条件变量（Condition）

条件变量允许线程在某些条件满足时进行等待，并在条件改变时唤醒等待的线程。它通常与锁一起使用。

示例：使用 threading.Condition 实现生产者-消费者模式

import threading

class Buffer:
    def __init__(self):
        self.buffer = []
        self.lock = threading.Lock()
        self.condition = threading.Condition(self.lock)

    def produce(self, item):
        with self.condition:
            while len(self.buffer) >= 5:  # 缓冲区已满
                self.condition.wait()  # 生产者等待
            self.buffer.append(item)
            print(f"生产者添加: {item}")
            self.condition.notify_all()  # 通知所有等待的消费者

    def consume(self):
        with self.condition:
            while len(self.buffer) == 0:  # 缓冲区为空
                self.condition.wait()  # 消费者等待
            item = self.buffer.pop(0)
            print(f"消费者获取: {item}")
            self.condition.notify_all()  # 通知所有等待的生产者

buffer = Buffer()

def producer():
    for i in range(10):
        buffer.produce(i)

def consumer():
    for _ in range(10):
        buffer.consume()

# 创建生产者和消费者线程
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

# 启动线程
producer_thread.start()
consumer_thread.start()

# 等待线程完成
producer_thread.join()
consumer_thread.join()

解释：

• condition = threading.Condition(lock)：创建一个条件变量对象，关联到一个锁。
• condition.wait()：使当前线程等待，直到其他线程调用 notify() 或 notify_all()。
• condition.notify_all()：唤醒所有等待该条件的线程。

4. 事件（Event）

事件是一种简单的线程间通信机制，允许一个线程等待某个特定事件的发生。

示例：使用 threading.Event 实现线程间的同步

import threading
import time

event = threading.Event()

def wait_for_event():
    print("线程正在等待事件触发...")
    event.wait()  # 阻塞，直到事件被设置
    print("事件已触发！")

def set_event():
    time.sleep(3)  # 模拟延迟
    event.set()  # 触发事件

# 创建并启动线程
wait_thread = threading.Thread(target=wait_for_event)
set_thread = threading.Thread(target=set_event)

wait_thread.start()
set_thread.start()

# 等待所有线程完成
wait_thread.join()
set_thread.join()

解释：

• event = threading.Event()：创建一个事件对象。
• event.wait()：阻塞当前线程，直到事件被设置。
• event.set()：设置事件，唤醒所有等待该事件的线程。