python线程间怎么通信

python线程间怎么通信 - 实践

Python 线程间通信（Inter-Thread Communication, ITC）是理解并发编程的核心之一。
由于同一进程内的所有线程共享内存空间，它比进程间通信（IPC）更高效，但也更危险 —— 稍不注意就会出现数据竞争、死锁等问题。

一、线程之间为什么要通信？

多线程的目的是让多个任务并行执行（例如 I/O 并发、后台计算、异步日志等）。
但这些线程往往需要共享数据或协调动作，比如：
• 一个线程生产数据，另一个线程消费；
• 主线程通知子线程停止；
• 多线程同时写日志；
• 多线程之间需要同步状态。

因此，需要安全的通信机制来传递数据、协调执行。

⸻

二、线程间通信的基本特征

请添加图片描述

⸻

⚙️ 三、常见的通信与同步机制

Python 提供了多种线程间通信方式，主要有以下几类：

请添加图片描述

⸻

1. Queue — 最推荐的通信方式

queue.Queue 是线程安全的 FIFO 队列（内部用锁同步），适合“生产者-消费者”模型。

示例：生产者与消费者

import threading, queue, time
q = queue.Queue(maxsize=5)
def producer():
for i in range(10):
item = f"任务-{i}"
q.put(item)      # 自动加锁
print(f"生产：{item}")
time.sleep(0.5)
def consumer():
while True:
item = q.get()   # 自动等待可用
print(f"消费：{item}")
time.sleep(1)
q.task_done()
threading.Thread(target=producer).start()
threading.Thread(target=consumer, daemon=True).start()

q.join() # 等所有任务完成

✅ 优点：
• 自动线程安全；
• 支持阻塞/非阻塞；
• 自带 join / task_done 协调机制；
• 可实现多生产者、多消费者模型。

⸻

2. Event — 信号通知机制

Event 是一种线程间事件信号：
一个线程等待事件，另一个线程触发事件。

示例

import threading, time
event = threading.Event()
def worker():
print("子线程等待信号...")
event.wait()          # 阻塞，直到被 set()
print("收到信号，继续执行！")
t = threading.Thread(target=worker)
t.start()
time.sleep(2)
print("主线程发信号")
event.set()               # 唤醒 worker
t.join()

✅ 优点：
• 实现线程之间的同步信号；
• 适合“启动/暂停/终止”等场景；
• 不需要共享数据。

⸻

3. Condition — 条件变量（适合复杂等待）

用于复杂的同步条件，比如：等待“队列非空”或“资源可用”。

示例

import threading, time
cond = threading.Condition()
data = []
def producer():
with cond:
print("生产者：准备数据")
data.append("任务数据")
cond.notify()  # 唤醒等待的消费者
def consumer():
with cond:
print("消费者：等待数据")
cond.wait()    # 等待 notify()
print("消费者：获取数据", data.pop())
threading.Thread(target=consumer).start()
time.sleep(1)
threading.Thread(target=producer).start()

✅ 特点：
• 可精确控制等待条件；
• 可配合多个线程使用；
• 是更底层的同步机制。

⸻

4. Lock / RLock — 最基础的同步方式

用于保护共享资源，防止并发修改冲突。

示例

import threading, time
lock = threading.Lock()
counter = 0
def worker():
global counter
for _ in range(100000):
with lock:
counter += 1
threads = [threading.Thread(target=worker) for _ in range(5)]
[t.start() for t in threads]
[t.join() for t in threads]
print("最终结果:", counter)

如果没有锁保护，多线程同时修改 counter，结果会错误。

✅ 区别：
• Lock：普通互斥锁；
• RLock：可重入锁（允许同一线程重复获得）。

⸻

5. Semaphore — 控制并发数量

类似计数锁，可用于限制并发资源访问数量。

示例

import threading, time
sem = threading.Semaphore(3)  # 最多允许3个线程同时执行
def worker(i):
with sem:
print(f"线程 {i} 开始")
time.sleep(2)
print(f"线程 {i} 结束")
for i in range(8):
threading.Thread(target=worker, args=(i,)).start()

✅ 应用：
• 限制数据库连接数；
• 控制文件下载并发数。

⸻

6. 全局变量（不推荐直接用）

线程共享全局变量，但需要锁保护。

示例

import threading
shared_data = 0
lock = threading.Lock()
def add():
global shared_data
for _ in range(100000):
with lock:
shared_data += 1
threads = [threading.Thread(target=add) for _ in range(5)]
[t.start() for t in threads]
[t.join() for t in threads]
print("结果:", shared_data)

⸻

7. Future / ThreadPoolExecutor（现代线程通信）

Python 的 concurrent.futures 封装了线程池和异步返回结果机制（Future 对象）。

示例

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def task(x):
time.sleep(1)
return x * x

with ThreadPoolExecutor(max_workers=4) as executor:
futures = [executor.submit(task, i) for i in range(6)]
for f in as_completed(futures):
print(“结果:”, f.result())

✅ 优点：
• 自带队列与锁；
• 线程安全；
• 结果异步返回（Future.result()）。

⸻

四、性能与适用场景对比

请添加图片描述

⸻

五、线程间通信的设计建议
1. ✅ 优先使用 Queue：最稳定、线程安全、简单。
2. ⚠️ 全局变量必须加锁：否则容易出现竞争条件。
3. 不要过度使用锁：过多锁会导致死锁、性能下降。
4. I/O 密集用多线程，CPU 密集用多进程。
5. 复杂同步逻辑用 Event / Condition 实现。
6. 任务分发场景可用 ThreadPoolExecutor。

⸻

✅ 六、总结对比表
请添加图片描述