一文彻底掌握Python线程编程

线程基础概念

线程是操作系统能够进行运算调度的最小单位，被包含在进程之中，是进程中的实际运作单位：

• 轻量级：创建线程比创建进程开销小
• 共享内存：同一进程内的线程共享内存空间
• 并发执行：多个线程可以同时运行
• I/O密集型优势：特别适合网络请求、文件读写等场景

创建和启动线程

Python通过threading模块提供线程操作功能。以下是创建线程的两种主要方法：

方法1：使用函数创建线程

import threading
import time

def print_numbers():
    for i in range(1, 6):
        time.sleep(0.5)
        print(f"Number: {i}")

def print_letters():
    for letter in ['A', 'B', 'C', 'D', 'E']:
        time.sleep(0.5)
        print(f"Letter: {letter}")

# 创建线程
t1 = threading.Thread(target=print_numbers)
t2 = threading.Thread(target=print_letters)

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

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

print("All threads completed!")

方法2：继承Thread类

import threading

class MyThread(threading.Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name
        
    def run(self):
        print(f"Thread {self.name} starting")
        # 线程执行的任务
        for i in range(3):
            print(f"{self.name}: {i}")
        print(f"Thread {self.name} ending")

# 创建并启动线程
threads = []
for i in range(3):
    t = MyThread(f"Thread-{i+1}")
    t.start()
    threads.append(t)

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

线程同步机制

当多个线程访问共享资源时，需要使用同步机制避免竞争条件：

使用Lock（锁）

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        # 获取锁
        lock.acquire()
        try:
            counter += 1
        finally:
            # 释放锁
            lock.release()

threads = []
for i in range(5):
    t = threading.Thread(target=increment)
    t.start()
    threads.append(t)

for t in threads:
    t.join()

print(f"Final counter value: {counter}")

使用Semaphore（信号量）

import threading
import time

# 限制同时访问资源的线程数为3
semaphore = threading.Semaphore(3)

def access_resource(thread_id):
    print(f"Thread {thread_id} is waiting to access")
    with semaphore:
        print(f"Thread {thread_id} has acquired the semaphore")
        time.sleep(2)
    print(f"Thread {thread_id} has released the semaphore")

threads = []
for i in range(10):
    t = threading.Thread(target=access_resource, args=(i,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()

理解GIL（全局解释器锁）

线程池最佳实践

使用线程池可以有效管理线程资源：

from concurrent.futures import ThreadPoolExecutor
import time

def task(n):
    print(f"Processing {n}")
    time.sleep(1)
    return n * n

# 创建线程池（最多4个线程）
with ThreadPoolExecutor(max_workers=4) as executor:
    # 提交任务
    futures = [executor.submit(task, i) for i in range(10)]
    
    # 获取结果
    results = [f.result() for f in futures]

print("Results:", results)

线程通信：使用Queue

import threading
import queue
import time

# 创建线程安全的队列
q = queue.Queue()

def producer():
    for i in range(5):
        time.sleep(0.5)
        item = f"Item {i}"
        q.put(item)
        print(f"Produced {item}")
    # 发送结束信号
    q.put(None)

def consumer():
    while True:
        item = q.get()
        if item is None:
            break
        print(f"Consumed {item}")
        q.task_done()

# 创建并启动线程
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

# 等待生产者完成
producer_thread.join()
# 等待消费者完成
consumer_thread.join()

print("All tasks completed")

Python线程适用场景