Python多线程编程实战：从GIL原理到树莓派传感器数据采集

1. 项目概述

在开发桌面应用或嵌入式系统（比如树莓派项目）时，我们经常会遇到一个头疼的问题：程序卡住了。比如，你的程序正在从网络下载一个大文件，或者从传感器慢悠悠地读取数据，整个用户界面就“冻”住了，点哪都没反应。又或者，你需要在后台处理数据的同时，前端还能实时响应用户的操作。这时候，多线程编程就成了解决问题的关键钥匙。

多线程允许你在一个程序里“同时”做多件事。注意，这里的“同时”打了引号，尤其是在Python的世界里，这背后有著名的全局解释器锁（GIL）在起作用，它使得多个线程无法真正并行执行CPU密集型任务（比如复杂的数学计算）。但对于I/O密集型任务——比如等待网络响应、读写文件、从GPIO口读取传感器数据——多线程的优势就非常明显了。当一个线程在等待I/O操作完成时，GIL会被释放，其他线程就可以获得执行权，从而极大地提升程序的整体响应速度和吞吐量。

本教程将聚焦于Python标准库中的threading模块，手把手教你如何从零开始创建和管理线程，并解决一个非常实际的问题：如何向线程函数传递参数。无论是向线程发送一个配置值，还是传递一个需要处理的数据列表，这都是构建实用多线程应用的基石。我们将通过具体的代码示例，从最简单的“Hello World”式线程，逐步深入到带参数传递的实用场景，并穿插讲解线程同步的基本概念和常见陷阱。无论你是想在树莓派上同时读取多个传感器数据，还是在桌面应用中构建一个响应迅捷的后台任务管理器，这里的内容都能为你提供扎实的起点。

2. 理解Python线程的核心：GIL与适用场景

在动手写代码之前，我们必须先搞清楚Python线程的“能力边界”，这能帮你避免选错工具，事倍功半。核心就在于理解全局解释器锁（Global Interpreter Lock, GIL）。

2.1 GIL是什么？为什么它如此重要？

你可以把GIL想象成一间只有一个麦克风的会议室（CPU核心）。Python解释器（CPython）规定，任何时候，只有一个线程（发言人）能拿着这个麦克风执行Python字节码。即使你的电脑有8个核心（8间会议室），但Python程序这个“大会”只被分配了一个麦克风。

这意味着：

• 对于CPU密集型任务：比如视频编码、大规模科学计算、复杂的循环处理。这类任务需要持续占用“麦克风”进行计算。多线程在这里几乎无用武之地，因为它们需要排队使用同一个麦克风，不仅不能加速，线程切换的开销反而可能让程序更慢。解决方案是使用multiprocessing（多进程）模块，每个进程有自己独立的解释器和内存空间，相当于有了多个会议室和多个麦克风，能实现真正的并行计算。
• 对于I/O密集型任务：比如网络请求（下载、API调用）、磁盘读写、串口通信、等待传感器数据。这类任务的特点是，线程在发起I/O请求后，大部分时间在等待外部设备响应。在等待期间，线程会主动释放GIL（放下麦克风），让其他线程有机会运行。这正是多线程大放异彩的地方。

2.2 多线程的典型应用场景

理解了GIL，我们就能精准地使用线程：

1. 提升GUI响应速度：在桌面应用中，将耗时的操作（如文件处理、网络查询）放入子线程，保证主线程（UI线程）始终流畅，避免界面“假死”。
2. 并发处理多个I/O操作：树莓派上同时读取温度、湿度传感器，并上传数据到云端。一个线程阻塞在传感器读取时，另一个线程可以处理网络发送。
3. 构建简单的服务器或消费者-生产者模型：使用一个线程监听网络连接，接收到请求后，将任务抛给另一个工作线程处理，实现简单的并发服务。

注意：多线程编程会引入共享状态的问题。多个线程同时读写同一个变量（如一个全局的计数器、一个列表）可能导致数据错乱，这就是竞态条件（Race Condition）。解决它需要用到锁（Lock）、**队列（Queue）**等同步原语，我们会在后续章节详细讨论。

3. 线程的创建、启动与等待：基础三板斧

让我们从最基础的开始。Python的threading模块使创建线程变得异常简单。

3.1 创建与启动一个简单线程

最基本的模式是：定义一个函数，然后将这个函数作为target参数传递给threading.Thread类。

import threading import time def my_task(): """这是一个将在新线程中运行的任务""" print(f"[子线程 {threading.current_thread().name}] 任务开始") time.sleep(2) # 模拟一个耗时2秒的I/O操作 print(f"[子线程 {threading.current_thread().name}] 任务结束") print(f"[主线程 {threading.current_thread().name}] 程序启动") # 1. 创建线程对象 # target参数指定要运行的函数，name参数可以给线程起个名字，方便调试 worker_thread = threading.Thread(target=my_task, name="WorkerThread-1") # 2. 启动线程 # 注意：不是直接调用my_task()，而是调用线程对象的start()方法 worker_thread.start() print(f"[主线程 {threading.current_thread().name}] 启动子线程后，我继续做我的事") time.sleep(0.5) # 主线程也稍微睡一下 print(f"[主线程 {threading.current_thread().name}] 主线程的其他工作也做完了")

运行这段代码，你会看到输出顺序可能是这样的：

[主线程 MainThread] 程序启动 [主线程 MainThread] 启动子线程后，我继续做我的事 [子线程 WorkerThread-1] 任务开始 [主线程 MainThread] 主线程的其他工作也做完了 [子线程 WorkerThread-1] 任务结束

关键点在于，worker_thread.start()之后，主线程并没有等待子线程结束，而是立刻继续执行后面的print语句。主线程和子线程是并发执行的。

3.2 使用join()进行线程同步

很多时候，我们需要主线程等待某个子线程完成工作后，才能继续。例如，主线程需要子线程的计算结果。这时就需要join()方法。

import threading import time def data_fetcher(): """模拟一个数据抓取任务""" print("[数据抓取线程] 开始抓取数据...") time.sleep(3) # 模拟网络延迟 fetched_data = "这是抓取到的数据" print("[数据抓取线程] 数据抓取完成！") return fetched_data # 注意：简单return，主线程是拿不到这个值的！后面会讲如何传递。 print("[主线程] 需要先获取数据才能进行下一步。") # 创建并启动抓取线程 fetch_thread = threading.Thread(target=data_fetcher, name="DataFetcher") fetch_thread.start() print("[主线程] 抓取线程已启动，现在等待它完成...") # 核心：调用join()，主线程在此阻塞，直到fetch_thread执行完毕 fetch_thread.join() print("[主线程] 抓取线程已完成，主线程继续执行后续处理...") # 这里理论上可以处理抓取到的数据了（虽然目前还拿不到）

join()的工作原理：调用join()的线程（通常是主线程）会进入阻塞状态，直到被join的那个线程终止。你可以把它理解为“线程汇合点”。

一个经典的应用场景：并行下载多个文件，等所有文件都下载完成后，再统一进行打包或处理。

def download_file(url, filename): # 模拟下载逻辑 time.sleep(random.uniform(0.5, 2.0)) print(f"{filename} 下载完成") urls = [('http://example.com/1.zip', 'file1.zip'), ('http://example.com/2.zip', 'file2.zip'), ('http://example.com/3.zip', 'file3.zip')] threads = [] for url, name in urls: t = threading.Thread(target=download_file, args=(url, name)) t.start() threads.append(t) # 保存线程对象 # 等待所有下载线程完成 for t in threads: t.join() print("所有文件下载完毕，开始打包...") # 执行打包逻辑

实操心得：join()可以设置一个超时参数，例如t.join(timeout=5.0)。这表示主线程最多等待5秒，超时后无论子线程是否完成都会继续执行。这在构建具有响应超时机制的程序时非常有用。

4. 向线程传递参数：args与kwargs详解

让线程执行一个固定函数意义有限，我们通常需要向线程函数传递动态的数据。threading.Thread()提供了两种方式：args和kwargs。

4.1 使用args传递位置参数

args接收一个元组（tuple），元组中的元素会按顺序传递给目标函数作为位置参数。

import threading def process_data(file_path, chunk_size, is_verbose): """模拟处理数据的函数，接收三个参数""" print(f"处理文件: {file_path}") print(f"块大小: {chunk_size}") if is_verbose: print("详细模式已开启") # ... 实际的数据处理逻辑 ... # 创建线程，并通过args传递参数 # 参数顺序必须与函数定义一致：(file_path, chunk_size, is_verbose) processor_thread = threading.Thread( target=process_data, args=("/path/to/data.csv", 1024, True) # 注意：这里是元组，单个元素后要加逗号 ) processor_thread.start() processor_thread.join()

关键细节：即使只传递一个参数，args也必须是一个元组。args=(“hello”, )是正确的（注意逗号），而args=(“hello”)就错了，这只是一个字符串。

4.2 使用kwargs传递关键字参数

kwargs接收一个字典（dict），字典的键值对会以关键字参数的形式传递给目标函数。这种方式代码可读性更高，尤其当参数很多时。

import threading def configure_device(ip_address, port=8080, timeout=10, retries=3): """配置一个网络设备，部分参数有默认值""" print(f"配置设备 -> IP: {ip_address}, 端口: {port}, 超时: {timeout}秒, 重试: {retries}次") # ... 配置逻辑 ... # 使用kwargs传递参数，顺序无关，且可以覆盖默认值 config_thread = threading.Thread( target=configure_device, kwargs={ 'ip_address': '192.168.1.100', 'timeout': 15, # 覆盖默认值10 # 'port' 使用默认值8080 # 'retries' 使用默认值3 } ) config_thread.start() config_thread.join()

4.3 args与kwargs混合使用

你也可以同时使用args和kwargs，但要注意，传给函数的参数不能重复或冲突。

def complex_task(name, priority, log_level='INFO', output_dir='./output'): print(f"任务: {name}, 优先级: {priority}, 日志级别: {log_level}, 输出目录: {output_dir}") # 混合传递：name通过args，其他通过kwargs task_thread = threading.Thread( target=complex_task, args=("后台数据备份", ), # name参数 kwargs={'priority': 5, 'output_dir': '/backup'} # 覆盖priority和output_dir的默认值 )

避坑指南：传递可变对象（如列表、字典）作为参数时需要格外小心。因为线程共享进程的内存空间，如果你在线程内部修改了传入的可变对象，这个修改在所有线程（包括主线程）中都是可见的。这可能是你期望的（用于共享数据），也可能导致意外的竞态条件。如果不想共享，可以在线程函数内部进行深拷贝（import copy; my_list = copy.deepcopy(passed_list)）。

5. 线程间的数据共享与通信：安全第一

线程创建了，参数也能传了，接下来就要面对多线程编程中最核心也最易出错的挑战：数据共享与同步。多个线程不加控制地读写同一块数据，就像几个人同时修改同一份文档而不告知对方，结果必然是混乱的。

5.1 使用锁（Lock）保护共享资源

锁是最基本的同步原语。它像是一个房间的钥匙，一次只允许一个线程持有钥匙进入房间（访问共享资源）。

import threading import time # 一个共享的计数器 shared_counter = 0 # 为这个计数器创建一把锁 counter_lock = threading.Lock() def increment_counter(iterations): global shared_counter # 声明使用全局变量 for _ in range(iterations): # 不安全的操作：直接修改 # shared_counter += 1 # 分解为：读取 -> 加1 -> 写回。这三步可能被其他线程打断。 # 安全的操作：使用锁 with counter_lock: # 使用with语句自动获取和释放锁 # 进入这个代码块的线程“持有锁”，其他线程会在此阻塞等待 current_value = shared_counter time.sleep(0.001) # 故意加入微小延迟，放大竞态条件 shared_counter = current_value + 1 # with块结束，锁自动释放 def unsafe_increment(iterations): global shared_counter for _ in range(iterations): current_value = shared_counter time.sleep(0.001) shared_counter = current_value + 1 # 测试不安全版本 print("--- 测试不安全版本 ---") shared_counter = 0 threads = [] for i in range(5): t = threading.Thread(target=unsafe_increment, args=(100,), name=f"Unsafe-{i}") threads.append(t) t.start() for t in threads: t.join() print(f"预期结果: 500, 实际结果: {shared_counter}") # 结果几乎肯定小于500 # 测试安全版本（使用锁） print("\n--- 测试安全版本（使用锁） ---") shared_counter = 0 threads = [] for i in range(5): t = threading.Thread(target=increment_counter, args=(100,), name=f"Safe-{i}") threads.append(t) t.start() for t in threads: t.join() print(f"预期结果: 500, 实际结果: {shared_counter}") # 结果总是500

锁的使用要点：

1. 粒度要细：锁保护的范围越小越好，只锁住真正共享的代码部分。长时间持有锁会降低并发性能。
2. 使用with语句：这是最推荐的方式，能确保锁在任何情况下（包括发生异常时）都会被释放，避免“死锁”。
3. 避免死锁：当线程A持有锁L1并请求锁L2，而线程B持有锁L2并请求锁L1时，两者都会永远等待，形成死锁。设计时应尽量避免嵌套锁，或按固定顺序获取锁。

5.2 使用队列（Queue）实现生产者-消费者模型

队列是线程间通信更安全、更高级的机制。queue.Queue是一个线程安全的FIFO（先进先出）队列，天然适合生产者-消费者模式：一些线程生产数据放入队列，另一些线程从队列取出数据消费。

import threading import queue import time import random # 创建一个线程安全的队列，最大容量为10 task_queue = queue.Queue(maxsize=10) def producer(producer_id): """生产者：生成任务放入队列""" for i in range(5): item = f"产品-P{producer_id}-{i}" # put()是阻塞的，如果队列满了，会等待直到有空位 task_queue.put(item) print(f"[生产者 {producer_id}] 生产了 {item}") time.sleep(random.random() * 0.5) # 模拟不稳定的生产速度 print(f"[生产者 {producer_id}] 生产完毕") def consumer(consumer_id): """消费者：从队列取出任务进行处理""" while True: try: # get()是阻塞的，如果队列为空，会等待直到有数据 # block=False 可设置非阻塞，queue.Empty异常 # timeout=2 可设置超时 item = task_queue.get(timeout=3) # 等待3秒，超时则退出 except queue.Empty: print(f"[消费者 {consumer_id}] 等待超时，退出") break print(f"[消费者 {consumer_id}] 正在处理 {item}") time.sleep(random.random() * 1) # 模拟处理耗时 print(f"[消费者 {consumer_id}] 处理完成 {item}") # 非常重要：通知队列该项任务已完成 task_queue.task_done() # 创建并启动生产者线程 producers = [] for i in range(2): p = threading.Thread(target=producer, args=(i,)) p.start() producers.append(p) # 创建并启动消费者线程 consumers = [] for i in range(3): c = threading.Thread(target=consumer, args=(i,)) c.start() consumers.append(c) # 等待所有生产者结束 for p in producers: p.join() # 等待队列中所有任务被消费完 task_queue.join() # 阻塞，直到队列中每个item都调用了task_done() # 此时队列已空，消费者线程会在get()超时后退出 # 等待所有消费者线程结束 for c in consumers: c.join() print("所有生产消费任务结束！")

队列的优势：

• 线程安全：put()和get()操作是原子的，内部已经实现了锁机制。
• 解耦：生产者和消费者不需要知道对方的存在，只与队列交互。
• 流量控制：通过maxsize可以控制队列容量，当队列满时，生产者会自动阻塞，防止内存被撑爆。

5.3 使用事件（Event）进行线程间通知

事件对象用于一个线程通知其他线程某个“事件”已经发生。一个线程等待事件，另一个线程触发事件。

import threading import time # 创建一个事件对象 start_event = threading.Event() data_ready_event = threading.Event() shared_data = None def worker_thread(): """工作线程，等待开始信号，然后处理数据""" print("[工作线程] 就绪，等待启动信号...") start_event.wait() # 阻塞，直到主线程调用 start_event.set() print("[工作线程] 收到启动信号！开始工作...") # 模拟工作 time.sleep(1) # 工作完成，准备好数据 global shared_data shared_data = "处理完成的结果" print("[工作线程] 数据处理完毕，通知主线程。") # 触发“数据就绪”事件 data_ready_event.set() def main_thread(): """主线程逻辑""" print("[主线程] 初始化系统...") time.sleep(2) print("[主线程] 发送启动信号给工作线程。") start_event.set() # 唤醒等待的工作线程 print("[主线程] 等待工作线程完成...") data_ready_event.wait() # 阻塞，直到工作线程调用 data_ready_event.set() print(f"[主线程] 收到通知。获取到的数据是: {shared_data}") if __name__ == "__main__": # 创建工作线程 worker = threading.Thread(target=worker_thread, name="Worker") worker.start() # 运行主线程逻辑（这里为了演示，也在一个线程里运行） main_thread() worker.join()

事件常用于：

• 线程初始化同步：主线程初始化完成后，通知所有工作线程开始。
• 优雅停止线程：设置一个stop_event，工作线程定期检查，主线程想停止时设置该事件。
• 协调多个线程的执行顺序。

6. 实战：树莓派多传感器数据采集案例

让我们结合一个树莓派上的实际场景，将前面所学串联起来。假设我们需要同时监测温度和湿度传感器（模拟I/O操作），并将数据定期上传到云端（另一个I/O操作）。

6.1 设计思路与架构

我们将创建三个线程：

1. 温度采集线程：每2秒读取一次温度数据，放入一个共享队列。
2. 湿度采集线程：每3秒读取一次湿度数据，放入同一个共享队列。
3. 数据上传线程：从队列中取出数据包（包含时间戳、传感器类型、值），打包并模拟上传到云端。

使用一个线程安全的队列作为数据缓冲区，解耦采集和上传。使用事件来控制程序的优雅停止。

6.2 代码实现

import threading import queue import time import random from datetime import datetime # 模拟停止事件，用于优雅关闭所有线程 stop_event = threading.Event() # 数据队列，用于在采集线程和上传线程之间传递数据 data_queue = queue.Queue(maxsize=50) def read_temperature(sensor_id): """模拟读取温度传感器（I/O操作）""" # 实际项目中，这里会是读取GPIO或I2C设备的代码，如： # import board, adafruit_dht # dht_device = adafruit_dht.DHT22(board.D4) # return dht_device.temperature time.sleep(0.05) # 模拟传感器读取的微小延迟 # 返回一个模拟的合理温度值 return round(random.uniform(20.0, 30.0), 2) def read_humidity(sensor_id): """模拟读取湿度传感器（I/O操作）""" # 实际代码类似温度读取 time.sleep(0.05) return round(random.uniform(40.0, 80.0), 2) def temperature_collector(collector_id, interval=2): """温度采集线程函数""" print(f"[温度采集器-{collector_id}] 启动，间隔{interval}秒") while not stop_event.is_set(): try: value = read_temperature(collector_id) timestamp = datetime.now().isoformat() data_packet = { 'timestamp': timestamp, 'sensor_type': 'temperature', 'sensor_id': collector_id, 'value': value, 'unit': '°C' } # 非阻塞方式放入队列，如果队列满则等待1秒 data_queue.put(data_packet, block=True, timeout=1) print(f"[温度采集器-{collector_id}] 采集到数据: {value}°C") except queue.Full: print(f"[温度采集器-{collector_id}] 警告：数据队列已满，丢弃一次数据") except Exception as e: print(f"[温度采集器-{collector_id}] 读取错误: {e}") # 等待指定间隔，但会检查停止事件 stop_event.wait(timeout=interval) print(f"[温度采集器-{collector_id}] 已停止") def humidity_collector(collector_id, interval=3): """湿度采集线程函数""" print(f"[湿度采集器-{collector_id}] 启动，间隔{interval}秒") while not stop_event.is_set(): try: value = read_humidity(collector_id) timestamp = datetime.now().isoformat() data_packet = { 'timestamp': timestamp, 'sensor_type': 'humidity', 'sensor_id': collector_id, 'value': value, 'unit': '%' } data_queue.put(data_packet, block=True, timeout=1) print(f"[湿度采集器-{collector_id}] 采集到数据: {value}%") except queue.Full: print(f"[湿度采集器-{collector_id}] 警告：数据队列已满，丢弃一次数据") except Exception as e: print(f"[湿度采集器-{collector_id}] 读取错误: {e}") stop_event.wait(timeout=interval) print(f"[湿度采集器-{collector_id}] 已停止") def data_uploader(uploader_id, batch_size=5): """数据上传线程函数""" print(f"[数据上传器-{uploader_id}] 启动") batch = [] while not stop_event.is_set() or not data_queue.empty(): try: # 从队列获取数据，超时1秒 packet = data_queue.get(timeout=1) batch.append(packet) data_queue.task_done() # 重要：标记任务完成 # 达到批次大小或停止时且队列为空时，上传一批数据 if len(batch) >= batch_size or (stop_event.is_set() and data_queue.empty() and batch): _upload_batch(uploader_id, batch) batch = [] except queue.Empty: # 队列为空是正常情况，继续循环 continue except Exception as e: print(f"[数据上传器-{uploader_id}] 上传错误: {e}") time.sleep(5) # 出错后等待一段时间再重试 # 循环结束后，上传剩余的数据 if batch: _upload_batch(uploader_id, batch) print(f"[数据上传器-{uploader_id}] 已停止") def _upload_batch(uploader_id, batch): """模拟上传一批数据到云端""" print(f"[数据上传器-{uploader_id}] 准备上传 {len(batch)} 条数据...") time.sleep(0.5) # 模拟网络延迟 # 这里应该是实际的HTTP POST请求，例如： # import requests # response = requests.post('https://api.example.com/data', json={'batch': batch}) print(f"[数据上传器-{uploader_id}] 批次上传成功。数据示例: {batch[0] if batch else '无'}") def main(): print("=== 树莓派多传感器数据采集系统启动 ===") # 创建并启动线程 temp_thread = threading.Thread(target=temperature_collector, args=(1,), name="TempCollector") humid_thread = threading.Thread(target=humidity_collector, args=(1,), name="HumidCollector") upload_thread = threading.Thread(target=data_uploader, args=(1,), name="DataUploader") temp_thread.start() humid_thread.start() upload_thread.start() # 主线程运行一段时间后，发出停止信号 try: print("系统运行中...按 Ctrl+C 停止。") time.sleep(15) # 让系统运行15秒 except KeyboardInterrupt: print("\n接收到中断信号。") finally: print("正在停止所有线程，请等待...") stop_event.set() # 设置停止事件 # 等待采集线程结束 temp_thread.join(timeout=3) humid_thread.join(timeout=3) # 等待上传线程处理完队列中剩余数据 upload_thread.join(timeout=5) print("=== 系统已安全停止 ===") if __name__ == "__main__": main()

6.3 案例解析与关键点

1. 线程函数设计：每个采集线程都是一个无限循环，通过stop_event.wait(timeout=interval)来实现定时采集，同时能及时响应停止信号。wait方法在超时或事件被设置时返回。
2. 优雅停止：使用stop_event是标准的线程停止模式。线程循环检查stop_event.is_set()，主程序想退出时调用stop_event.set()，所有线程会在完成当前迭代后退出。
3. 异常处理：在put和get操作中加入了异常处理（queue.Full,queue.Empty），并处理了可能的传感器读取错误，增强了程序的健壮性。
4. 数据打包：将数据封装成字典（JSON格式），包含时间戳、类型、值、单位，便于后续处理和上传。
5. 流量控制与背压：队列设置最大容量（maxsize=50），当采集速度超过上传速度导致队列满时，put操作会短暂阻塞或超时，丢弃数据并告警，这形成了简单的背压机制，防止内存无限增长。

7. 常见问题、调试技巧与性能考量

即使掌握了基本用法，在实际开发中你仍会遇到各种问题。这里记录了一些典型的坑和解决思路。

7.1 常见问题排查表

7.2 调试与监控技巧

• 给线程命名：创建线程时使用name参数（如Thread(target=..., name=”SensorReader”)），在日志和调试信息中能清晰区分。
• 使用日志模块：替代print，使用logging模块，可以输出线程名、时间戳和日志级别，便于追踪。

  import logging logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def worker(): logger.info(“线程开始工作”)

• 使用threading.enumerate()：在任意位置调用，可以列出所有存活的线程对象，用于检查线程状态。
• 使用threading.current_thread()：在线程函数内调用，获取当前线程对象，可访问其name、ident等属性。

7.3 性能考量与进阶选择

• 线程池（ThreadPoolExecutor）：对于需要执行大量短期异步任务的场景（如处理大量HTTP请求），频繁创建销毁线程开销大。使用concurrent.futures.ThreadPoolExecutor可以管理一个可复用的线程池。

  from concurrent.futures import ThreadPoolExecutor, as_completed def fetch_url(url): # 模拟网络请求 time.sleep(1) return f”Data from {url}” urls = [“url1”, “url2”, “url3”, “url4”] with ThreadPoolExecutor(max_workers=3) as executor: # 最多3个线程并发 future_to_url = {executor.submit(fetch_url, url): url for url in urls} for future in as_completed(future_to_url): url = future_to_url[future] try: data = future.result() print(f”{url} 返回: {data}”) except Exception as exc: print(f”{url} 产生异常: {exc}”)

• 异步IO（asyncio）：对于超大规模（数千上万）的并发网络连接，线程的上下文切换开销会成为瓶颈。Python的asyncio提供了基于事件循环的单线程并发模型，在I/O密集型场景下资源利用率更高，性能更好。如果你的项目主要是网络通信，且Python版本在3.5+，强烈建议学习asyncio。
• 多进程（multiprocessing）：如前所述，对于CPU密集型任务（如图像处理、数据计算），必须使用multiprocessing模块来绕过GIL，利用多核CPU。进程间通信（IPC）比线程间通信开销大，常用multiprocessing.Queue或multiprocessing.Pipe。

多线程是Python并发编程工具箱中不可或缺的一件利器，尤其擅长解决I/O阻塞导致的程序响应性问题。从简单的Thread(target=func)开始，到使用Lock、Queue、Event构建复杂的线程协作应用，关键在于理解共享状态的风险并妥善管理它。在树莓派这类资源受限但I/O操作频繁的平台上，合理运用多线程能极大提升项目的实用性和用户体验。记住，如果遇到CPU瓶颈，别忘了还有multiprocessing这个兄弟模块。在实际项目中，多结合日志和调试工具，从小功能开始实践，你会逐渐掌握这门让程序“同时”做多件事的艺术。