如何使用 Python 线程

简介

本全面教程将探讨Python线程，为开发者提供利用并发编程技术的基本知识。通过理解线程基础、同步机制和实际实现策略，程序员可以创建更高效、响应更快的Python应用程序，从而有效利用多核处理器。

线程基础

什么是线程？

线程是一种编程技术，它允许程序的多个部分在单个进程中并发运行。在Python中，threading 模块提供了一种创建和管理线程的方法，从而实现代码的并行执行。

为什么要使用线程？

线程在以下场景中特别有用：

你需要同时执行多个任务
某些任务涉及I/O操作或等待
你希望提高程序的整体性能

场景	线程的好处
网页抓取	并行数据收集
网络编程	处理多个连接
CPU密集型任务	可能提高性能

基本线程创建

以下是在Python中创建和运行线程的简单示例：

import threading
import time

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

## 创建多个线程
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("所有线程已完成")

线程生命周期

stateDiagram-v2 [*] --> 创建创建 --> 可运行可运行 --> 运行运行 --> 阻塞阻塞 --> 可运行运行 --> 终止终止 --> [*]

线程类型

守护线程：后台线程，不会阻止程序退出
非守护线程：线程会使程序一直运行，直到它们完成

线程安全注意事项

线程共享相同的内存空间
对共享资源的并发访问可能导致竞态条件
正确的同步至关重要

性能注意事项

虽然线程可以提高性能，但并不总是最佳解决方案：

Python的全局解释器锁（GIL）限制了真正的并行执行
对于CPU密集型任务，考虑使用多进程
I/O密集型任务从线程中受益最大

LabEx提示

在学习线程时，LabEx提供了实践和试验线程编程技术的实践环境。

要避免的常见陷阱

过度使用线程
忽略同步
创建过多线程
线程终止不当

线程同步

理解线程同步

线程同步是一种控制对共享资源的访问并防止多线程应用程序中出现竞态条件的机制。

同步机制

1. 锁（互斥锁）

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, iterations):
    for _ in range(iterations):
        counter.increment()

## 演示线程安全的递增操作
counter = Counter()
threads = []
for _ in range(5):
    t = threading.Thread(target=worker, args=(counter, 1000))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"最终计数器值: {counter.value}")

2. 可重入锁（递归锁）

import threading

class RecursiveLockExample:
    def __init__(self):
        self.rlock = threading.RLock()

    def method1(self):
        with self.rlock:
            print("方法1获取了锁")
            self.method2()

    def method2(self):
        with self.rlock:
            print("方法2获取了锁")

同步原语

原语	描述	使用场景
锁（Lock）	基本的互斥	简单的临界区
可重入锁（RLock）	可重入的锁	嵌套锁获取
信号量（Semaphore）	限制并发访问	资源池
事件（Event）	线程间的信号传递	协调
条件变量（Condition）	高级等待机制	复杂的同步

信号量示例

import threading
import time

class LimitedResourcePool:
    def __init__(self, max_connections):
        self.semaphore = threading.Semaphore(max_connections)

    def acquire_resource(self, thread_id):
        self.semaphore.acquire()
        try:
            print(f"线程 {thread_id} 获取了资源")
            time.sleep(2)
        finally:
            self.semaphore.release()
            print(f"线程 {thread_id} 释放了资源")

def worker(pool, thread_id):
    pool.acquire_resource(thread_id)

## 演示信号量的使用
resource_pool = LimitedResourcePool(max_connections=2)
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(resource_pool, i))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

同步流程

sequenceDiagram participant Thread1 participant SharedResource participant Thread2 Thread1->>SharedResource: 获取锁 Thread2->>SharedResource: 等待锁 Thread1-->>SharedResource: 修改资源 Thread1->>SharedResource: 释放锁 Thread2->>SharedResource: 获取锁

常见同步挑战

死锁
竞态条件
优先级反转

最佳实践

尽量减少临界区
使用最简单的同步机制
尽可能避免嵌套锁
注意性能开销

LabEx建议

在LabEx的交互式Python环境中练习线程同步技术，以获得实践经验。

潜在陷阱

过度同步可能导致性能瓶颈
不正确的锁管理可能导致死锁
复杂的同步逻辑可能引入难以调试的错误

实际线程示例

使用并发线程进行网页抓取

import threading
import requests
from queue import Queue

class WebScraper:
    def __init__(self, urls):
        self.urls = urls
        self.results = {}
        self.queue = Queue()
        self.lock = threading.Lock()

    def fetch_url(self):
        while not self.queue.empty():
            url = self.queue.get()
            try:
                response = requests.get(url, timeout=5)
                with self.lock:
                    self.results[url] = len(response.text)
            except Exception as e:
                with self.lock:
                    self.results[url] = str(e)
            finally:
                self.queue.task_done()

    def scrape(self, max_threads=5):
        for url in self.urls:
            self.queue.put(url)

        threads = []
        for _ in range(max_threads):
            t = threading.Thread(target=self.fetch_url)
            t.start()
            threads.append(t)

        self.queue.join()
        return self.results

## 示例用法
urls = [
    'https://www.example.com',
    'https://www.python.org',
    'https://www.github.com'
]
scraper = WebScraper(urls)
results = scraper.scrape()
print(results)

并行文件处理

import os
import threading
from concurrent.futures import ThreadPoolExecutor

class FileProcessor:
    def __init__(self, directory):
        self.directory = directory
        self.processed_files = []
        self.lock = threading.Lock()

    def process_file(self, filename):
        file_path = os.path.join(self.directory, filename)
        try:
            with open(file_path, 'r') as f:
                content = f.read()
                processed_content = content.upper()

            with self.lock:
                self.processed_files.append({
                    'filename': filename,
                    'size': len(processed_content)
                })
        except Exception as e:
            print(f"处理 {filename} 时出错: {e}")

    def process_files(self, max_workers=4):
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            files = [f for f in os.listdir(self.directory) if os.path.isfile(os.path.join(self.directory, f))]
            executor.map(self.process_file, files)

        return self.processed_files

使用事件进行线程通信

import threading
import time

class TrafficLight:
    def __init__(self):
        self.green_light = threading.Event()
        self.red_light = threading.Event()

    def traffic_controller(self):
        while True:
            ## 绿灯
            self.green_light.set()
            self.red_light.clear()
            print("绿灯 - 交通通行")
            time.sleep(5)

            ## 红灯
            self.green_light.clear()
            self.red_light.set()
            print("红灯 - 交通停止")
            time.sleep(3)

    def vehicle(self, name):
        while True:
            if self.green_light.is_set():
                print(f"{name} 正在通过")
            else:
                print(f"{name} 正在等待")
            time.sleep(1)

## 示例用法
traffic = TrafficLight()
controller = threading.Thread(target=traffic.traffic_controller)
controller.daemon = True
controller.start()

vehicles = []
for i in range(3):
    v = threading.Thread(target=traffic.vehicle, args=(f"车辆-{i}",))
    v.daemon = True
    v.start()
    vehicles.append(v)

## 保持主线程运行
for v in vehicles:
    v.join()