推荐10篇常用的多线程编用法

《推荐10篇常用的多线程编程用法》

多线程编程是Python中提升程序性能的重要手段，尤其在处理I/O密集型任务或需要并行计算的场景下。本文将通过10个经典案例，系统讲解Python中多线程的核心用法，涵盖基础线程创建、线程同步、线程池管理、线程安全等关键技术点，帮助读者快速掌握多线程编程的实践技巧。

1. 基础线程创建：threading.Thread

Python通过threading模块提供线程支持，最基础的用法是继承Thread类或直接实例化并传入目标函数。

import threading
import time

def worker(num):
    print(f"线程{num}开始执行")
    time.sleep(2)
    print(f"线程{num}执行结束")

# 创建并启动线程
threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

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

print("所有线程执行完毕")

关键点：通过start()启动线程，join()阻塞主线程等待子线程结束。这种方式适合简单任务，但需手动管理线程生命周期。

2. 线程同步：Lock锁机制

当多个线程共享资源时，需使用锁（Lock）避免竞态条件。以下示例展示线程安全的数据操作：

import threading

class Counter:
    def __init__(self):
        self.value = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:  # 自动获取和释放锁
            self.value += 1

def worker(counter, num):
    for _ in range(1000):
        counter.increment()
    print(f"线程{num}完成计数")

counter = Counter()
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(counter, i))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"最终计数结果: {counter.value}")  # 应为5000

关键点：with语句确保锁的自动释放，避免死锁。锁的粒度需合理，过粗会影响性能，过细则可能失效。

3. 线程间通信：Event对象

Event是线程间通信的简单方式，通过设置/清除标志位控制线程执行流程：

import threading
import time

def waiter(event):
    print("等待事件触发...")
    event.wait()  # 阻塞直到event被set()
    print("事件已触发，继续执行")

def notifier(event):
    time.sleep(3)
    print("触发事件")
    event.set()  # 唤醒所有等待线程

event = threading.Event()
t1 = threading.Thread(target=waiter, args=(event,))
t2 = threading.Thread(target=notifier, args=(event,))

t1.start()
t2.start()

t1.join()
t2.join()

应用场景：主线程通知子线程继续执行，或子线程等待特定条件满足。

4. 线程池：concurrent.futures

手动管理线程开销大，线程池（ThreadPoolExecutor）可复用线程资源，提升效率：

from concurrent.futures import ThreadPoolExecutor
import time

def task(name):
    print(f"任务{name}开始")
    time.sleep(2)
    return f"任务{name}完成"

with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交多个任务
    futures = [executor.submit(task, i) for i in range(5)]
    # 获取结果（按完成顺序）
    for future in concurrent.futures.as_completed(futures):
        print(future.result())

优势：自动管理线程生命周期，支持结果回调和超时控制，适合批量任务处理。

5. 线程安全队列：Queue模块

Queue提供线程安全的FIFO队列，常用于生产者-消费者模型：

import threading
import queue
import time

def producer(q):
    for i in range(5):
        print(f"生产数据: {i}")
        q.put(i)
        time.sleep(1)

def consumer(q):
    while True:
        item = q.get()
        if item is None:  # 终止信号
            break
        print(f"消费数据: {item}")
        q.task_done()

q = queue.Queue()
t1 = threading.Thread(target=producer, args=(q,))
t2 = threading.Thread(target=consumer, args=(q,))

t1.start()
t2.start()

t1.join()
q.put(None)  # 发送终止信号
t2.join()

关键方法：put()、get()、task_done()、join()（阻塞直到所有任务完成）。

6. 定时线程：Timer类

Timer用于延迟执行任务，适合定时任务场景：

from threading import Timer

def hello():
    print("Hello, World!")

# 5秒后执行
t = Timer(5.0, hello)
t.start()
print("定时器已启动，等待执行...")

# 可通过t.cancel()取消定时任务

注意事项：Timer是单次执行，如需周期性任务，需在回调中重新启动Timer。

7. 守护线程：daemon属性

守护线程（daemon）在主线程退出时自动终止，适合后台任务：

import threading
import time

def background_task():
    while True:
        print("后台线程运行中...")
        time.sleep(1)

t = threading.Thread(target=background_task)
t.daemon = True  # 设置为守护线程
t.start()

print("主线程退出后，守护线程将终止")
time.sleep(3)  # 主线程休眠3秒后退出

关键点：必须在start()前设置daemon=True，否则无效。

8. 线程局部变量：local()

每个线程拥有独立的变量副本，避免数据冲突：

import threading

local_data = threading.local()

def worker():
    local_data.value = threading.current_thread().name
    print(f"{threading.current_thread().name}的value: {local_data.value}")

threads = []
for i in range(3):
    t = threading.Thread(target=worker, name=f"Thread-{i}")
    threads.append(t)
    t.start()

for t in threads:
    t.join()

应用场景：Web服务器中存储线程独立的会话数据。

9. 线程优先级（模拟）

Python原生不支持线程优先级，但可通过调整睡眠时间模拟：

import threading
import time

def task(priority, name):
    wait_time = 1.0 / (priority + 1)  # 优先级越高，等待时间越短
    time.sleep(wait_time)
    print(f"优先级{priority}的{name}执行完毕")

threads = []
for prio, name in [(1, "高"), (3, "中"), (5, "低")]:
    t = threading.Thread(target=task, args=(prio, name))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

替代方案：使用multiprocessing或第三方库如priority实现真实优先级。

10. 多线程与GIL的权衡

Python的GIL（全局解释器锁）限制了多线程的CPU密集型任务性能，此时应改用多进程：

from multiprocessing import Pool
import time

def cpu_bound(x):
    return sum(i*i for i in range(1000000))

if __name__ == "__main__":
    # 多线程（受GIL限制）
    start = time.time()
    threads = []
    for _ in range(4):
        t = threading.Thread(target=lambda: cpu_bound(0))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print(f"多线程耗时: {time.time()-start:.2f}秒")

    # 多进程（无GIL限制）
    start = time.time()
    with Pool(4) as p:
        p.map(cpu_bound, [0]*4)
    print(f"多进程耗时: {time.time()-start:.2f}秒")

结论：I/O密集型用多线程，CPU密集型用多进程。

总结与最佳实践

1. 优先使用concurrent.futures简化线程管理
2. 共享数据时务必加锁或使用线程安全容器
3. 合理设置线程池大小（通常为CPU核心数*2~3）
4. 避免在子线程中操作GUI等主线程资源
5. 监控线程状态，及时处理异常

关键词：Python多线程、threading模块、线程同步、线程池、Queue队列、守护线程、线程局部变量、GIL限制、多进程替代、生产者消费者模型

简介：本文系统介绍Python多线程编程的10种核心用法，涵盖线程创建、同步、通信、线程池、安全队列等关键技术，结合代码示例解析实际应用场景，并对比多线程与多进程的适用条件，帮助开发者高效处理并发任务。