Python 线程创建全攻略：从入门到精通

多线程是 Python 中实现并发编程的重要方式，它可以让你同时执行多个任务，提高程序效率。本文将全面介绍 Python 中创建线程的各种方法。

一、为什么需要多线程？

在开始之前，先了解多线程的适用场景：

• I/O 密集型任务：网络请求、文件读写、数据库操作
• 用户界面响应：保持界面流畅的同时执行后台任务
• 并发处理：同时处理多个客户端请求
• 定时任务：定期执行某些操作

二、线程的基本概念

1. 线程 vs 进程

特性	线程	进程
资源开销	小	大
创建速度	快	慢
内存共享	共享同一进程内存	独立内存空间
通信方式	直接共享变量	需要 IPC 机制

2. Python 的 GIL（全局解释器锁）

需要注意的是，由于 GIL 的存在，Python 的多线程不适合 CPU 密集型任务。对于 CPU 密集型任务，建议使用多进程。

三、创建线程的 4 种方法

方法1：使用 threading.Thread 类（最常用）

import threading
import time

def print_numbers(name, count):
    """线程任务函数：打印数字"""
    for i in range(1, count + 1):
        print(f"{name}: {i}")
        time.sleep(0.1)

# 创建线程
thread1 = threading.Thread(target=print_numbers, args=("线程A", 5))
thread2 = threading.Thread(target=print_numbers, args=("线程B", 5))

# 启动线程
thread1.start()
thread2.start()

# 等待线程结束
thread1.join()
thread2.join()

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

方法2：继承 Thread 类

import threading
import time

class MyThread(threading.Thread):
    def __init__(self, name, count):
        super().__init__()
        self.name = name
        self.count = count
    
    def run(self):
        """重写 run 方法，定义线程执行的任务"""
        for i in range(1, self.count + 1):
            print(f"{self.name}: {i}")
            time.sleep(0.1)
        print(f"{self.name} 执行完成")

# 创建并启动线程
threads = []
for i in range(3):
    thread = MyThread(f"自定义线程-{i}", 3)
    thread.start()
    threads.append(thread)

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

print("所有自定义线程执行完成")

方法3：使用线程池（推荐）

from concurrent.futures import ThreadPoolExecutor
import time

def task(name, duration):
    """线程任务"""
    print(f"{name} 开始执行，耗时 {duration}秒")
    time.sleep(duration)
    print(f"{name} 执行完成")
    return f"{name}-结果"

# 创建线程池（最大3个线程）
with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交任务
    future1 = executor.submit(task, "任务1", 2)
    future2 = executor.submit(task, "任务2", 1)
    future3 = executor.submit(task, "任务3", 3)
    future4 = executor.submit(task, "任务4", 1)
    
    # 获取结果
    results = [future1.result(), future2.result(), future3.result(), future4.result()]
    print("所有任务结果:", results)

方法4：使用 threading.Timer 定时线程

import threading

def delayed_task(message):
    print(f"定时任务执行: {message}")

# 创建定时器（3秒后执行）
timer = threading.Timer(3.0, delayed_task, args=("Hello World!",))
print("定时器已设置，3秒后执行...")
timer.start()

# 可以取消定时器
# timer.cancel()

四、线程同步与通信

1. 使用锁（Lock）防止资源竞争

import threading

class BankAccount:
    def __init__(self, balance=0):
        self.balance = balance
        self.lock = threading.Lock()
    
    def deposit(self, amount):
        """存款（线程安全）"""
        with self.lock:  # 自动获取和释放锁
            print(f"存款前余额: {self.balance}")
            self.balance += amount
            print(f"存款 {amount}，当前余额: {self.balance}")
    
    def withdraw(self, amount):
        """取款（线程安全）"""
        with self.lock:
            if self.balance >= amount:
                print(f"取款前余额: {self.balance}")
                self.balance -= amount
                print(f"取款 {amount}，当前余额: {self.balance}")
            else:
                print("余额不足")

# 测试
account = BankAccount(1000)

def customer_operations(name, operations):
    for op_type, amount in operations:
        if op_type == "deposit":
            account.deposit(amount)
        else:
            account.withdraw(amount)
        threading.Event().wait(0.1)  # 稍微延迟

# 创建多个线程模拟并发操作
threads = []
threads.append(threading.Thread(target=customer_operations, 
                               args=("客户A", [("deposit", 200), ("withdraw", 100)])))
threads.append(threading.Thread(target=customer_operations,
                               args=("客户B", [("withdraw", 300), ("deposit", 150)])))

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

print(f"最终余额: {account.balance}")

2. 使用队列（Queue）进行线程间通信

import threading
import queue
import time
import random

# 生产者-消费者模式
def producer(q, name):
    """生产者线程"""
    for i in range(5):
        item = f"{name}-产品{i}"
        print(f"生产者 {name} 生产: {item}")
        q.put(item)
        time.sleep(random.uniform(0.1, 0.5))
    q.put(None)  # 结束信号

def consumer(q, name):
    """消费者线程"""
    while True:
        item = q.get()
        if item is None:
            q.put(None)  # 传递给其他消费者
            break
        print(f"消费者 {name} 消费: {item}")
        time.sleep(random.uniform(0.2, 0.8))
        q.task_done()

# 创建队列
q = queue.Queue()

# 创建生产者和消费者线程
producers = [
    threading.Thread(target=producer, args=(q, "工厂A")),
    threading.Thread(target=producer, args=(q, "工厂B"))
]

consumers = [
    threading.Thread(target=consumer, args=(q, "消费者1")),
    threading.Thread(target=consumer, args=(q, "消费者2"))
]

# 启动所有线程
for p in producers:
    p.start()

for c in consumers:
    c.start()

# 等待生产者完成
for p in producers:
    p.join()

# 等待队列清空
q.join()

# 等待消费者完成
for c in consumers:
    c.join()

print("生产消费完成")

五、线程实战案例

案例1：并发下载器

import threading
import requests
import time
from concurrent.futures import ThreadPoolExecutor

def download_file(url, filename):
    """下载文件"""
    start_time = time.time()
    try:
        response = requests.get(url, timeout=10)
        with open(filename, 'wb') as f:
            f.write(response.content)
        end_time = time.time()
        print(f"下载完成: {filename} (耗时: {end_time - start_time:.2f}s)")
        return True
    except Exception as e:
        print(f"下载失败: {filename}, 错误: {e}")
        return False

# 要下载的文件列表
urls = [
    ("https://example.com/file1.jpg", "file1.jpg"),
    ("https://example.com/file2.jpg", "file2.jpg"),
    ("https://example.com/file3.jpg", "file3.jpg"),
    ("https://example.com/file4.jpg", "file4.jpg")
]

# 使用线程池并发下载
start_time = time.time()

with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交所有下载任务
    futures = [executor.submit(download_file, url, filename) 
              for url, filename in urls]
    
    # 等待所有任务完成
    results = [future.result() for future in futures]

end_time = time.time()
print(f"总耗时: {end_time - start_time:.2f}s")
print(f"成功下载: {sum(results)}/{len(results)} 个文件")

案例2：Web 服务器请求处理

import threading
import time
import random

class WebServer:
    def __init__(self):
        self.request_count = 0
        self.lock = threading.Lock()
    
    def handle_request(self, client_id):
        """处理客户端请求"""
        # 模拟请求处理时间
        processing_time = random.uniform(0.1, 1.0)
        time.sleep(processing_time)
        
        with self.lock:
            self.request_count += 1
            current_count = self.request_count
        
        print(f"请求 #{current_count:03d} 来自客户端 {client_id} "
              f"(处理时间: {processing_time:.2f}s)")

# 创建服务器实例
server = WebServer()

# 模拟并发客户端请求
clients = range(1, 21)  # 20个客户端
threads = []

for client_id in clients:
    thread = threading.Thread(target=server.handle_request, args=(client_id,))
    threads.append(thread)
    thread.start()
    # 稍微错开请求时间
    time.sleep(random.uniform(0, 0.1))

# 等待所有请求处理完成
for thread in threads:
    thread.join()

print(f"\n总共处理了 {server.request_count} 个请求")

六、线程最佳实践

1. 使用线程池而不是创建大量线程

# 不推荐：创建大量线程
threads = []
for i in range(1000):
    thread = threading.Thread(target=some_task)
    thread.start()
    threads.append(thread)

# 推荐：使用线程池
with ThreadPoolExecutor(max_workers=50) as executor:
    results = executor.map(some_task, range(1000))

2. 正确处理异常

def safe_task():
    try:
        # 可能出错的代码
        result = risky_operation()
        return result
    except Exception as e:
        print(f"任务执行失败: {e}")
        return None

# 使用
thread = threading.Thread(target=safe_task)
thread.start()

3. 使用守护线程

def background_task():
    while True:
        print("后台任务运行中...")
        time.sleep(5)

# 创建守护线程（主程序退出时自动结束）
daemon_thread = threading.Thread(target=background_task, daemon=True)
daemon_thread.start()

# 主程序
time.sleep(10)
print("主程序结束，守护线程会自动终止")

4. 避免死锁

# 错误的做法：嵌套获取多个锁
lock1 = threading.Lock()
lock2 = threading.Lock()

def risky_function():
    with lock1:
        with lock2:  # 可能死锁
            # 操作共享资源
            pass

# 正确的做法：按固定顺序获取锁
def safe_function():
    with lock1:
        # 操作只需要lock1的资源
        pass
    
    with lock2:
        # 操作只需要lock2的资源
        pass

七、常见问题与解决方案

1. 线程间数据共享

import threading
from threading import local

# 使用线程局部存储
thread_local = local()

def show_thread_data():
    # 每个线程有自己的数据
    if hasattr(thread_local, 'data'):
        print(f"线程 {threading.current_thread().name}: {thread_local.data}")
    else:
        print(f"线程 {threading.current_thread().name}: 没有数据")

def worker(name):
    thread_local.data = f"数据来自 {name}"
    show_thread_data()

# 测试
threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(f"线程{i}",))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

2. 线程超时控制

import threading
import time

def long_running_task():
    print("任务开始")
    time.sleep(10)  # 模拟长时间运行
    print("任务完成")
    return "结果"

# 创建线程
thread = threading.Thread(target=lambda: print(long_running_task()))

# 启动线程并设置超时
thread.start()
thread.join(timeout=3)  # 等待3秒

if thread.is_alive():
    print("任务超时，强制终止")
    # 注意：强制终止线程可能不安全
else:
    print("任务正常完成")

八、总结

Python 线程编程要点：

• 选择合适的方法：根据需求选择 Thread 类、继承方式或线程池
• 理解 GIL 限制：线程适合 I/O 密集型任务，不适合 CPU 密集型任务
• 重视线程安全：使用锁、队列等机制保护共享资源
• 使用线程池：避免创建过多线程，提高资源利用率
• 处理异常：确保线程异常不会导致整个程序崩溃
• 合理通信：使用队列等安全方式进行线程间通信