MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Python中使用ThreadPoolExecutor进行线程管理

2021-07-154.3k 阅读

Python中的线程管理基础

在Python编程领域，多线程编程是一项强大的技术，它允许我们同时执行多个任务，从而提高程序的效率和响应性。线程管理是多线程编程的核心部分，涉及到线程的创建、启动、暂停、终止以及资源分配等操作。Python提供了多种线程管理工具，其中ThreadPoolExecutor是一个非常实用的模块，它基于concurrent.futures库，为我们提供了一个高级的异步执行框架，使得线程管理变得更加简单和高效。

线程的概念

在深入探讨ThreadPoolExecutor之前，我们先来回顾一下线程的基本概念。线程是操作系统能够进行运算调度的最小单位，它被包含在进程之中，是进程中的实际运作单位。一个进程可以包含多个线程，这些线程共享进程的资源，如内存空间、文件描述符等。

多线程编程的优势在于能够充分利用多核CPU的性能，提高程序的执行效率。例如，在一个网络爬虫程序中，我们可以使用多线程同时下载多个网页，从而加快数据采集的速度。然而，多线程编程也带来了一些挑战，比如线程安全问题、资源竞争等，这些问题需要我们在编写代码时特别注意。

Python的线程模块

Python提供了多个线程相关的模块，其中最常用的是threading模块。threading模块允许我们创建和管理线程，通过继承threading.Thread类或者直接调用threading.Thread构造函数来创建线程对象。以下是一个简单的threading模块使用示例：

import threading


def worker():
    print('Worker thread is running')


t = threading.Thread(target=worker)
t.start()
t.join()

在这个示例中，我们定义了一个worker函数，然后通过threading.Thread创建了一个线程对象t，并将worker函数作为目标函数传递给线程对象。接着，我们调用start方法启动线程，最后调用join方法等待线程执行完毕。

虽然threading模块提供了基本的线程管理功能，但在处理大量线程或者需要更高级的异步执行特性时，它可能会显得有些繁琐。这时候，ThreadPoolExecutor就派上用场了。

ThreadPoolExecutor简介

ThreadPoolExecutor是concurrent.futures库中的一个类，它提供了一个线程池，可以管理一组线程并异步执行任务。线程池是一种资源池技术，它预先创建一定数量的线程，并将这些线程放入线程池中。当有任务需要执行时，线程池会从池中取出一个空闲线程来执行任务，任务执行完毕后，线程并不会销毁，而是返回线程池等待下一个任务。这种方式避免了频繁创建和销毁线程带来的开销，提高了程序的性能。

导入ThreadPoolExecutor

在使用ThreadPoolExecutor之前，我们需要先导入它。ThreadPoolExecutor位于concurrent.futures库中，因此我们可以使用以下方式导入：

from concurrent.futures import ThreadPoolExecutor

创建ThreadPoolExecutor对象

创建ThreadPoolExecutor对象非常简单，我们只需要指定线程池中的最大线程数即可。例如，要创建一个最大线程数为5的线程池，可以使用以下代码：

executor = ThreadPoolExecutor(max_workers=5)

这里的max_workers参数指定了线程池中的最大线程数。如果不指定该参数，ThreadPoolExecutor会根据系统的CPU核心数自动选择一个合适的默认值。

使用ThreadPoolExecutor执行任务

ThreadPoolExecutor提供了两种主要的方法来提交任务并获取结果：submit和map。

submit方法

submit方法用于提交一个任务到线程池并返回一个Future对象。Future对象代表一个异步执行的任务，可以用来查询任务的状态、获取任务的结果或者取消任务。以下是一个使用submit方法的示例：

import concurrent.futures
import time


def task(x):
    print(f'Starting task {x}')
    time.sleep(2)
    print(f'Finishing task {x}')
    return x * x


with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    future = executor.submit(task, 5)
    try:
        result = future.result(timeout=3)
        print(f'Task result: {result}')
    except concurrent.futures.TimeoutError:
        print('Task timed out')

在这个示例中，我们定义了一个task函数，它接收一个参数x，模拟一个需要执行2秒的任务，并返回x的平方。然后，我们使用ThreadPoolExecutor创建了一个最大线程数为3的线程池，并通过submit方法提交了一个任务到线程池。接着，我们使用future.result方法获取任务的结果，并设置了超时时间为3秒。如果任务在3秒内没有完成，将会抛出TimeoutError异常。

map方法

map方法类似于Python内置的map函数，它可以将一个函数应用到多个参数上，并返回一个迭代器，该迭代器按顺序返回每个任务的结果。map方法会自动为每个参数创建一个任务并提交到线程池执行。以下是一个使用map方法的示例：

import concurrent.futures
import time


def task(x):
    print(f'Starting task {x}')
    time.sleep(2)
    print(f'Finishing task {x}')
    return x * x


with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(task, range(5)))
    print(f'Task results: {results}')

在这个示例中，我们同样定义了task函数，然后使用ThreadPoolExecutor的map方法将task函数应用到range(5)生成的0到4这5个参数上。map方法会自动为每个参数创建一个任务并提交到线程池执行，最后返回一个迭代器，我们将其转换为列表并打印出每个任务的结果。

Future对象详解

在使用ThreadPoolExecutor的submit方法提交任务时，会返回一个Future对象。Future对象提供了一些方法来查询任务的状态、获取任务的结果、取消任务等。

查询任务状态

Future对象提供了几个方法来查询任务的状态：

  • running()：如果任务正在执行且尚未完成，返回True。
  • done()：如果任务已完成（无论是正常完成还是发生异常），返回True。
  • cancelled()：如果任务已被取消，返回True。

以下是一个示例：

import concurrent.futures
import time


def task():
    time.sleep(2)
    return 42


with concurrent.futures.ThreadPoolExecutor() as executor:
    future = executor.submit(task)
    while not future.done():
        print('Task is still running...')
        time.sleep(1)
    print('Task is done')

在这个示例中，我们通过while循环不断检查future.done()的返回值，直到任务完成。

获取任务结果

我们可以使用result()方法获取任务的结果。如果任务尚未完成，result()方法会阻塞当前线程，直到任务完成。我们还可以设置一个超时时间，如果在超时时间内任务没有完成，会抛出TimeoutError异常。

import concurrent.futures
import time


def task():
    time.sleep(2)
    return 42


with concurrent.futures.ThreadPoolExecutor() as executor:
    future = executor.submit(task)
    try:
        result = future.result(timeout=3)
        print(f'Task result: {result}')
    except concurrent.futures.TimeoutError:
        print('Task timed out')

取消任务

Future对象的cancel()方法可以尝试取消任务。如果任务尚未开始执行，取消操作会成功，cancel()方法返回True；如果任务已经开始执行，取消操作可能会失败，cancel()方法返回False。

import concurrent.futures
import time


def task():
    time.sleep(5)
    return 42


with concurrent.futures.ThreadPoolExecutor() as executor:
    future = executor.submit(task)
    cancelled = future.cancel()
    if cancelled:
        print('Task was cancelled')
    else:
        print('Task could not be cancelled')

在这个示例中，由于任务需要执行5秒，而我们在任务提交后立即尝试取消任务，此时任务尚未开始执行，所以取消操作会成功。

线程池的高级应用

动态调整线程池大小

在实际应用中，我们可能需要根据任务的负载动态调整线程池的大小。虽然ThreadPoolExecutor本身并没有直接提供动态调整线程池大小的方法，但我们可以通过一些技巧来实现类似的功能。

一种方法是创建多个不同大小的线程池，并根据任务的数量或者执行时间来选择合适的线程池。例如，我们可以根据任务队列的长度来决定是否需要创建一个更大的线程池：

import concurrent.futures
import time


def task(x):
    time.sleep(1)
    return x * x


task_queue = []
for i in range(10):
    task_queue.append(i)

if len(task_queue) > 5:
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
else:
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)

results = list(executor.map(task, task_queue))
print(f'Task results: {results}')

在这个示例中，我们根据任务队列的长度来决定使用一个最大线程数为5还是2的线程池。

处理异常

在多线程编程中，任务执行过程中可能会发生异常。ThreadPoolExecutor提供了一种简单的方式来处理这些异常。当任务发生异常时，Future对象的result()方法会抛出异常，我们可以在捕获异常时进行相应的处理。

import concurrent.futures


def task(x):
    if x < 0:
        raise ValueError('Input must be non - negative')
    return x * x


with concurrent.futures.ThreadPoolExecutor() as executor:
    futures = [executor.submit(task, i) for i in range(-2, 3)]
    for future in concurrent.futures.as_completed(futures):
        try:
            result = future.result()
            print(f'Task result: {result}')
        except ValueError as e:
            print(f'Caught exception: {e}')

在这个示例中，我们定义的task函数在输入为负数时会抛出ValueError异常。我们使用as_completed函数来迭代已完成的任务，并在获取任务结果时捕获可能发生的异常。

线程池与I/O操作

ThreadPoolExecutor特别适合处理I/O密集型任务，如网络请求、文件读写等。因为在I/O操作过程中，线程大部分时间处于等待状态，此时可以让其他线程继续执行任务，从而提高程序的整体效率。

以下是一个使用ThreadPoolExecutor进行网络请求的示例：

import concurrent.futures
import requests


def fetch_url(url):
    response = requests.get(url)
    return response.status_code


urls = [
    'http://www.example.com',
    'http://www.google.com',
    'http://www.github.com'
]

with concurrent.futures.ThreadPoolExecutor() as executor:
    results = list(executor.map(fetch_url, urls))
    for url, status_code in zip(urls, results):
        print(f'{url}: {status_code}')

在这个示例中，我们定义了fetch_url函数来发送HTTP请求并获取响应状态码。通过ThreadPoolExecutor的map方法，我们可以同时发送多个网络请求，大大提高了效率。

线程安全与资源竞争

在多线程编程中，线程安全和资源竞争是必须要考虑的问题。由于多个线程共享进程的资源，当多个线程同时访问和修改共享资源时，可能会导致数据不一致或者程序崩溃。

锁机制

Python提供了锁（Lock）机制来解决资源竞争问题。锁是一种同步原语，它可以保证在同一时间只有一个线程能够访问共享资源。threading模块中的Lock类可以用于创建锁对象。

以下是一个使用锁来保护共享资源的示例：

import threading


counter = 0
lock = threading.Lock()


def increment():
    global counter
    with lock:
        counter = counter + 1


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

for t in threads:
    t.join()

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

在这个示例中，我们定义了一个共享变量counter和一个锁对象lock。在increment函数中，我们使用with lock语句来获取锁，这样在锁的作用域内，只有当前线程能够访问和修改counter变量，从而避免了资源竞争。

线程本地存储

除了锁机制，线程本地存储（Thread - Local Storage，TLS）也是一种解决线程安全问题的方法。TLS允许每个线程拥有自己独立的变量副本，这样不同线程之间就不会相互干扰。

在Python中，threading.local()函数可以创建一个线程本地对象。以下是一个示例：

import threading


local_data = threading.local()


def set_local_value(value):
    local_data.value = value
    print(f'Set local value to {local_data.value} in thread {threading.current_thread().name}')


def get_local_value():
    try:
        print(f'Get local value {local_data.value} in thread {threading.current_thread().name}')
    except AttributeError:
        print('Local value not set in thread {threading.current_thread().name}')


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

for t in threads:
    t.join()

for i in range(3):
    t = threading.Thread(target = get_local_value)
    threads.append(t)
    t.start()

在这个示例中，我们通过threading.local()创建了一个local_data对象。每个线程可以独立地设置和获取local_data.value，不同线程之间的local_data.value相互独立，从而避免了资源竞争。

与其他并发编程模型的比较

与进程的比较

线程和进程都是实现并发编程的方式，但它们有一些重要的区别。进程是操作系统资源分配的基本单位，每个进程都有自己独立的内存空间和系统资源。而线程是进程中的执行单元，多个线程共享进程的资源。

使用线程的优点是线程之间的切换开销比进程小，适合处理I/O密集型任务。但由于线程共享资源，可能会引发资源竞争问题。进程则更适合处理CPU密集型任务，因为每个进程可以充分利用多核CPU的性能，并且进程之间相互独立，不存在资源竞争问题。

与异步I/O的比较

异步I/O是另一种实现并发编程的方式，它通过事件循环和回调函数来处理I/O操作，避免了线程切换的开销。在Python中，asyncio库提供了异步I/O的支持。

与线程相比，异步I/O更适合处理高并发的I/O密集型任务，因为它不需要创建大量线程，从而减少了资源消耗。但异步I/O的编程模型相对复杂，需要使用async和await关键字来定义异步函数和等待异步操作完成。而线程编程模型则更直观，对于一些简单的并发任务，使用线程可能更容易理解和实现。

综上所述，ThreadPoolExecutor为Python开发者提供了一种简单而强大的线程管理方式，适用于各种I/O密集型任务和部分CPU密集型任务。通过合理使用线程池、处理线程安全问题以及与其他并发编程模型进行比较和选择，我们可以编写出高效、稳定的并发程序。在实际应用中，需要根据具体的需求和场景来选择最合适的并发编程方式。