Python 多线程编程技巧

用户11081884

发布于 2026-07-20 18:52:31

110

Python中的多线程编程是一种常见的并发编程方式，特别适合I/O密集型任务。虽然由于GIL(全局解释器锁)的存在，Python的多线程并不能实现真正的并行计算，但在处理网络请求、文件操作等I/O密集型任务时，多线程仍然能显著提高程序效率。

基本线程创建

方法一：使用threading.Thread

import threading
import time

def download_file(url):
    print(f"开始下载 {url}")
    time.sleep(2)  # 模拟下载耗时
    print(f"完成下载 {url}")

# 创建线程
thread1 = threading.Thread(target=download_file, args=("http://example.com/file1.zip",))
thread2 = threading.Thread(target=download_file, args=("http://example.com/file2.zip",))

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

# 等待线程完成
thread1.join()
thread2.join()

print("所有下载任务完成")

方法二：继承Thread类

import threading
import time

class DownloadThread(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
    
    def run(self):
        print(f"开始下载 {self.url}")
        time.sleep(2)  # 模拟下载耗时
        print(f"完成下载 {self.url}")

# 创建并启动线程
threads = [
    DownloadThread("http://example.com/file1.zip"),
    DownloadThread("http://example.com/file2.zip"),
    DownloadThread("http://example.com/file3.zip")
]

for t in threads:
    t.start()

for t in threads:
    t.join()

print("所有下载任务完成")

线程池的使用

线程池是管理多线程的更高效方式，可以避免频繁创建和销毁线程的开销。

from concurrent.futures import ThreadPoolExecutor
import time

def process_data(data):
    print(f"处理数据: {data}")
    time.sleep(1)  # 模拟处理耗时
    return f"处理结果-{data}"

# 使用线程池处理数据
with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交多个任务
    futures = [executor.submit(process_data, i) for i in range(10)]
    
    # 获取结果
    for future in futures:
        print(future.result())

print("所有数据处理完成")

线程同步与锁机制

当多个线程需要访问共享资源时，需要使用锁来保证数据一致性。

import threading

class BankAccount:
    def __init__(self):
        self.balance = 1000
        self.lock = threading.Lock()
    
    def deposit(self, amount):
        with self.lock:
            self.balance += amount
            print(f"存入 {amount}, 余额: {self.balance}")
    
    def withdraw(self, amount):
        with self.lock:
            if self.balance >= amount:
                self.balance -= amount
                print(f"取出 {amount}, 余额: {self.balance}")
            else:
                print(f"余额不足, 当前余额: {self.balance}")

account = BankAccount()

def customer_operation(operations):
    for op, amount in operations:
        if op == "deposit":
            account.deposit(amount)
        else:
            account.withdraw(amount)

# 创建两个线程模拟两个客户操作账户
thread1 = threading.Thread(
    target=customer_operation,
    args=([("deposit", 200), ("withdraw", 500), ("deposit", 300)],)
)

thread2 = threading.Thread(
    target=customer_operation,
    args=([("withdraw", 800), ("deposit", 1000), ("withdraw", 200)],)
)

thread1.start()
thread2.start()

thread1.join()
thread2.join()

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

线程间通信

使用队列(queue.Queue)实现线程间安全的数据交换。

import threading
import queue
import time
import random

def producer(q, items):
    for item in items:
        time.sleep(random.uniform(0.1, 0.5))  # 模拟生产时间
        q.put(item)
        print(f"生产: {item}")
    # 所有产品生产完后，为每个消费者发送结束信号
    for _ in range(3):
        q.put(None)

def consumer(q, name):
    while True:
        item = q.get()
        if item is None:        # 收到结束信号
            print(f"{name} 结束")
            q.task_done()
            break
        time.sleep(random.uniform(0.2, 0.8))  # 模拟消费时间
        print(f"{name} 消费: {item}")
        q.task_done()

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

# 生产数据
items = [f"产品-{i}" for i in range(10)]

# 创建生产者线程
producer_thread = threading.Thread(target=producer, args=(q, items))

# 创建消费者线程
consumers = [
    threading.Thread(target=consumer, args=(q, f"消费者-{i}"))
    for i in range(3)
]

# 启动所有线程
producer_thread.start()
for c in consumers:
    c.start()

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

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

# 等待所有消费者结束
for c in consumers:
    c.join()

print("所有任务完成")

实用技巧

1、合理设置线程数量：I/O密集型任务可以设置较多线程，但不要过多，避免上下文切换开销。

2、使用线程局部数据：threading.local()可以为每个线程保存独立的数据。

import threading

local_data = threading.local()

def show_data():
    try:
        print(f"线程 {threading.current_thread().name} 的数据: {local_data.value}")
    except AttributeError:
        print(f"线程 {threading.current_thread().name} 没有数据")

def worker(value):
    local_data.value = value
    show_data()

threads = [
    threading.Thread(target=worker, args=(i,))
    for i in range(3)
]

for t in threads:
    t.start()

for t in threads:
    t.join()

3、使用定时器线程：threading.Timer可以实现延迟执行。

import threading

def delayed_action(message):
    print(message)

# 3 秒后执行
timer = threading.Timer(3.0, delayed_action, args=("3 秒后执行的任务",))
timer.start()

print("主线程继续执行...")

4、使用事件(Event)协调线程：

import threading
import time

event = threading.Event()

def waiter():
    print("等待事件触发...")
    event.wait()
    print("事件已触发，继续执行")

def setter():
    time.sleep(2)
    print("准备设置事件")
    event.set()

t1 = threading.Thread(target=waiter)
t2 = threading.Thread(target=setter)

t1.start()
t2.start()

t1.join()
t2.join()

Python多线程编程虽然受到GIL的限制，但在I/O密集型任务中仍然非常有用。掌握线程创建、线程池、线程同步和线程间通信等技巧，可以显著提高程序的并发性能：