登录
首页 >  文章 >  python教程

Python多线程教程：threading模块详解

时间：2025-11-04 19:41:34 390浏览 收藏

Python多线程编程是提升I/O密集型任务效率的关键技术。本文**详解Python threading模块的使用**，包括**创建线程的两种方法**：传递可调用对象和继承Thread类，并通过示例代码展示如何启动、管理线程以及等待线程完成。同时，深入探讨了**多线程与多进程的选择**，强调了GIL对CPU密集型任务的限制，并给出了针对不同任务类型的优化建议。此外，文章还介绍了**Lock和RLock**，用于解决多线程数据同步与竞争条件的问题，以及**线程池ThreadPoolExecutor和守护线程**，以优化线程生命周期和资源管理，助你掌握Python多线程编程的核心技巧。

Python多线程通过threading模块实现，适用于I/O密集型任务，利用线程提升并发效率；尽管受GIL限制无法在CPU密集型任务中并行执行，但结合Lock/RLock可解决共享资源竞争问题，而ThreadPoolExecutor和守护线程则优化了线程生命周期与资源管理。

python如何进行多线程编程_python threading模块多线程实现方法

Python多线程编程主要通过内置的threading模块实现。它允许程序创建和管理独立的执行流，从而在某些场景下提升程序的并发能力，尤其是在处理I/O密集型任务时。尽管受限于全局解释器锁（GIL），它在CPU密集型任务中并非真正的并行，但对于等待网络响应、文件读写等操作，threading依然是提升效率的有效工具。

解决方案

要使用Python的threading模块实现多线程，核心是创建Thread对象并启动它们。你可以通过两种主要方式来定义线程的执行逻辑：

  1. 传递一个可调用对象（函数）给Thread构造函数： 这是最直接、最常见的方法。

    import threading
import time

def task_function(name, delay):
    """一个简单的线程任务函数"""
    print(f"线程 {name}: 启动")
    time.sleep(delay)
    print(f"线程 {name}: 完成")

# 创建线程实例
thread1 = threading.Thread(target=task_function, args=("Thread-1", 2))
thread2 = threading.Thread(target=task_function, args=("Thread-2", 3))

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

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

print("主线程: 所有子线程已完成。")

    target参数接收线程要执行的函数，args参数是一个元组，包含传递给target函数的参数。

  2. 继承threading.Thread类并重写run()方法： 这种方式更适合需要封装线程逻辑和状态的场景。

    import threading
import time

class MyThread(threading.Thread):
    def __init__(self, name, delay):
        super().__init__()
        self.name = name
        self.delay = delay

    def run(self):
        """线程执行的实际逻辑"""
        print(f"线程 {self.name}: 启动")
        time.sleep(self.delay)
        print(f"线程 {self.name}: 完成")

# 创建线程实例
thread1 = MyThread("MyThread-1", 2)
thread2 = MyThread("MyThread-2", 3)

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

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

print("主线程: 所有自定义线程已完成。")

    无论哪种方式，start()方法都会启动线程，使其调用run()方法（或target函数）。join()方法则会阻塞当前线程（通常是主线程），直到对应的子线程执行完毕。这对于确保所有后台任务都完成，或者需要等待子线程结果的场景非常关键。

Python多线程与多进程：何时选择，如何权衡？

关于Python多线程和多进程的选择，这确实是初学者常常困惑的地方。我个人经验是，这主要取决于你的任务类型和对“并行”的理解。Python的全局解释器锁（GIL）是绕不开的话题，它保证了在任何给定时刻，只有一个线程能执行Python字节码。这意味着，即使你启动了多个线程，它们也无法在多核CPU上真正地同时执行CPU密集型任务。

所以，如果你的程序主要是I/O密集型（比如网络请求、文件读写、数据库操作），这意味着程序大部分时间都在等待外部资源响应，CPU是空闲的。在这种情况下，threading模块就非常有用了。当一个线程在等待I/O时，GIL会被释放，允许其他线程运行，从而有效利用等待时间，提升程序的并发处理能力。我记得有一次处理大量图片下载，使用多线程后，速度提升了好几倍，因为大部分时间都在等待图片从服务器传输过来。

然而，如果你的程序是CPU密集型（比如复杂的数学计算、图像处理、数据分析），那么threading几乎不会带来性能提升，甚至可能因为线程切换的开销而略微降低性能。这时候，multiprocessing模块才是正解。它通过创建独立的进程来绕过GIL，每个进程都有自己的Python解释器和内存空间，因此可以在多核CPU上实现真正的并行执行。当然，进程间的通信和数据共享会比线程复杂一些，开销也更大。

总结来说，我的建议是：

  • I/O密集型任务：优先考虑threading，它更轻量，数据共享相对简单。
  • CPU密集型任务：果断选择multiprocessing，以获得真正的并行性能。
  • 混合型任务：可以考虑结合使用，例如在一个进程中启动多个线程来处理I/O，或者使用进程池来分发CPU密集型任务。

多线程数据同步与竞争条件：Lock和RLock的实践

在多线程编程中，当多个线程尝试同时修改或访问同一个共享资源时，就可能出现所谓的“竞争条件”（Race Condition）。这会导致数据不一致或程序行为异常，是多线程编程中最常见的陷阱之一。我记得有一次，就是因为忘记加锁，导致一个计数器在并发环境下总是得到错误的结果，那真是个让人头疼的bug，排查了很久才发现是共享变量的问题。

为了避免这种情况，threading模块提供了一系列同步原语，其中最基础和常用的是Lock（锁）。

threading.Lock

Lock是一个互斥锁（Mutex），它确保在任何给定时刻，只有一个线程可以持有锁。当一个线程acquire()了锁之后，其他尝试acquire()该锁的线程会被阻塞，直到持有锁的线程release()它。

import threading
import time

shared_counter = 0
lock = threading.Lock() # 创建一个锁

def increment_counter():
    global shared_counter
    for _ in range(100000):
        # 使用with语句管理锁，确保即使发生异常也能释放锁
        with lock:
            shared_counter += 1

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

for thread in threads:
    thread.join()

print(f"最终计数器值: {shared_counter}") # 期望值是 5 * 100000 = 500000

如果没有lock，shared_counter的最终值几乎不可能是500000，因为多个线程会同时读取旧值、递增，然后写回，导致一些递增操作被覆盖。with lock:语法是推荐的使用方式，它会自动在代码块开始时acquire()锁，并在代码块结束时（无论正常结束还是发生异常）release()锁，非常安全。

threading.RLock (可重入锁)

RLock是“可重入锁”，它和Lock类似，但允许同一个线程多次acquire()同一个锁，只要该线程之前已经持有该锁。每次acquire()都需要对应一次release()。RLock主要用于防止死锁，特别是当一个函数需要调用另一个也需要获取相同锁的函数时。

import threading

r_lock = threading.RLock()

def func_a():
    with r_lock:
        print(f"{threading.current_thread().name} 进入 func_a")
        func_b()
        print(f"{threading.current_thread().name} 退出 func_a")

def func_b():
    with r_lock:
        print(f"{threading.current_thread().name} 进入 func_b")
        # 模拟一些操作
        print(f"{threading.current_thread().name} 退出 func_b")

thread = threading.Thread(target=func_a, name="MyReentrantThread")
thread.start()
thread.join()

在这个例子中，func_a获取了r_lock，然后它又调用了func_b，func_b再次尝试获取r_lock。如果这里用的是Lock，就会发生死锁，因为func_a已经持有锁，func_b无法再次获取。但使用RLock，同一个线程可以多次获取，只要释放次数与获取次数匹配即可。

选择Lock还是RLock取决于你的具体需求。如果你的代码结构可能导致同一个线程需要多次获取同一个锁，那么RLock是更安全的选项。否则，Lock通常就足够了。

线程池与守护线程：优化多线程程序的管理

在实际的多线程应用中，我们往往需要更精细地管理线程的生命周期和资源消耗。手动创建和销毁大量线程会带来不小的开销，而且如果线程数量失控，还可能耗尽系统资源。这时候，线程池和守护线程的概念就显得尤为重要。

守护线程 (Daemon Threads)

守护线程是一种特殊类型的线程，它的生命周期与主线程密切相关。当你将一个线程设置为守护线程（通过thread.daemon = True），意味着当所有非守护线程（包括主线程）都结束时，即使守护线程还没有完成其任务，Python解释器也会强制终止它们。

import threading
import time

def daemon_task():
    print("守护线程: 启动，开始后台工作...")
    time.sleep(5) # 模拟长时间运行
    print("守护线程: 完成工作。")

def non_daemon_task():
    print("非守护线程: 启动，执行短期任务...")
    time.sleep(1)
    print("非守护线程: 完成任务。")

daemon_thread = threading.Thread(target=daemon_task)
daemon_thread.daemon = True # 设置为守护线程

non_daemon_thread = threading.Thread(target=non_daemon_task)

daemon_thread.start()
non_daemon_thread.start()

non_daemon_thread.join() # 等待非守护线程完成

print("主线程: 非守护线程已完成，主线程即将退出。")
# 此时，如果daemon_thread还没完成，它也会被强制终止。

守护线程非常适合那些在后台默默运行、提供辅助服务（如日志记录、缓存清理、心跳检测）的任务，它们不需要等待主程序退出后才结束。但要注意，由于它们可能被突然终止，所以不适合执行那些需要保证数据完整性或资源正确释放的任务。

线程池 (Thread Pool)

对于那些需要处理大量短期任务的场景，线程池简直是救星。它让代码更整洁，也更有效率，不用自己手动管理那么多线程的生老病死。线程池预先创建了一定数量的线程，并将它们放入一个池子中。当有新任务到来时，线程池会从池中取出一个空闲线程来执行任务；任务完成后，线程不会被销毁，而是返回池中等待下一个任务。这样就避免了频繁创建和销毁线程的开销，也限制了同时运行的线程数量，防止资源过度消耗。

Python标准库中的concurrent.futures模块提供了ThreadPoolExecutor，这是实现线程池的现代且推荐的方式。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def process_data(item):
    """模拟一个耗时的数据处理任务"""
    print(f"正在处理数据: {item}...")
    time.sleep(2) # 模拟处理时间
    result = f"处理完成: {item} -> 结果"
    print(f"处理结果: {result}")
    return result

data_items = [f"Item-{i}" for i in range(10)]

# 创建一个最大工作线程数为3的线程池
with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交任务到线程池，并获取Future对象
    futures = [executor.submit(process_data, item) for item in data_items]

    print("\n等待所有任务完成并获取结果:")
    # 使用as_completed迭代已完成的Future
    for future in as_completed(futures):
        try:
            result = future.result() # 获取任务的返回值
            print(f"从Future获取结果: {result}")
        except Exception as exc:
            print(f"任务生成了一个异常: {exc}")

print("\n所有任务已提交并处理完毕。")

ThreadPoolExecutor的submit()方法会提交一个任务并返回一个Future对象，你可以通过Future.result()获取任务的返回值，或者通过as_completed()迭代已完成的Future对象，这让异步结果的处理变得非常优雅。线程池极大地简化了并发任务的管理，是构建高性能、高并发应用的利器。

今天带大家了解了的相关知识，希望对你有所帮助；关于文章的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~

相关阅读
更多>
最新阅读
更多>
课程推荐
更多>