Python并发编程：高效获取最快完成任务的结果

本文详细阐述了在python并发编程中，如何高效地启动多个任务并仅获取其中最快完成任务的结果，同时忽略其他耗时任务。通过引入`concurrent.futures`模块，特别是`threadpoolexecutor`和`as_completed`方法，我们能够以简洁且非阻塞的方式实现这一目标，极大简化了线程管理和结果检索的复杂性，适用于需要快速响应的场景，如处理多个api请求。

Python并发编程中获取最快任务结果的策略

在进行并发编程时，我们经常会遇到这样的场景：需要同时执行多个独立的任务，但我们只关心其中最快完成的那一个任务的结果，而对于其他耗时较长的任务，我们希望能够直接忽略它们，并继续执行后续逻辑。这种需求在处理多个API请求、执行多种计算策略或从多个数据源获取信息时尤为常见，目标是最小化总等待时间。

传统threading模块的局限性

Python标准库中的threading模块提供了创建和管理线程的基本功能。然而，直接使用threading.Thread来解决“获取最快完成任务结果”的问题会面临一些挑战：

1. 结果返回机制不直接： threading.Thread的target函数执行完毕后，其返回值无法直接通过线程对象获取。通常需要借助共享数据结构（如队列、列表或自定义类属性）并配合锁机制来传递结果，增加了复杂性。
2. 阻塞式等待： 如果要等待线程完成并获取结果，通常需要调用thread.join()。但join()是阻塞的，它会等待特定线程完全执行完毕。如果我们需要获取 最快 完成的线程结果，就无法简单地对所有线程都调用join()，因为这会使程序等待最慢的线程。
3. 事件管理复杂： 尝试使用threading.Event等机制来通知主线程某个任务已完成，虽然可行，但需要手动编写复杂的同步逻辑，且难以优雅地获取具体任务的返回值。

考虑以下使用threading模块的尝试，它无法实现预期效果：

import threading, time

def one():
    time.sleep(1)
    return 1

def two():
    time.sleep(5)
    return 2

thread_one = threading.Thread(target=one)
thread_two = threading.Thread(target=two)

thread_one.start()
thread_two.start()

# 这里的 print(one()) 会直接调用函数 one()，而不是获取线程 one 的结果
# 并且它会阻塞主线程 1 秒，与并发的目的相悖。
# 实际上，这里并没有从线程中获取结果的机制。
print(one())

上述代码中的print(one())并不是从已启动的线程中获取结果，而是再次同步地调用了one()函数，导致主线程阻塞1秒，并且没有利用到线程并发的优势。

立即学习“Python免费学习笔记（深入）”；

解决方案：concurrent.futures模块

Python 3 引入的concurrent.futures模块提供了一种更高级、更抽象的并发执行接口，它封装了线程和进程的创建与管理，使得编写并发代码变得更加简单和安全。对于获取最快完成任务结果的需求，concurrent.futures中的ThreadPoolExecutor（或ProcessPoolExecutor）结合as_completed方法是理想的选择。

ThreadPoolExecutor：线程池管理

ThreadPoolExecutor是一个线程池执行器，它维护一个线程池，可以向其提交任务。提交任务后，执行器会从池中选择一个可用线程来执行任务。这避免了手动创建和管理线程的开销。

MindShow

MindShow官网 | AI生成PPT，快速演示你的想法

1492

查看详情

submit()：提交任务

executor.submit(fn, *args, **kwargs)方法用于向执行器提交一个可调用对象（函数）及其参数。它会立即返回一个Future对象。Future对象代表了任务的未来结果，它是一个占位符，你可以在任务完成后通过它获取结果、检查状态或捕获异常。

as_completed()：按完成顺序迭代

concurrent.futures.as_completed(futures, timeout=None)是解决我们问题的关键。它接收一个Future对象的可迭代集合，并返回一个迭代器。这个迭代器会按照任务完成的顺序，逐个产生已经完成的Future对象。这意味着，一旦有任务完成，as_completed就会立即返回对应的Future，而无需等待所有任务都完成。

Future.result()：获取任务结果

当从as_completed迭代器中获取到一个Future对象后，可以调用其result()方法来获取任务的实际返回值。如果任务尚未完成，result()方法会阻塞直到任务完成。但由于as_completed已经保证了返回的Future是已完成的，所以调用result()通常会立即返回结果（或抛出异常）。

示例代码与解释

下面是使用concurrent.futures模块实现获取最快完成任务结果的示例：

import concurrent.futures
import time

# 定义两个模拟任务，耗时不同
def one():
    time.sleep(1)  # 模拟耗时1秒
    print("任务 one 完成")
    return 1

def two():
    time.sleep(5)  # 模拟耗时5秒
    print("任务 two 完成")
    return 2

# 使用 with 语句创建 ThreadPoolExecutor，确保线程池在任务完成后被正确关闭
with concurrent.futures.ThreadPoolExecutor() as pool:
    # 提交任务到线程池，并获取对应的 Future 对象
    tasks = [
        pool.submit(one),
        pool.submit(two),
    ]

    print("所有任务已提交，等待最快任务完成...")

    # as_completed 会在任务完成时逐个返回 Future 对象
    # next() 用于获取第一个完成的 Future 对象
    first_completed_future = next(concurrent.futures.as_completed(tasks))

    # 从第一个完成的 Future 对象中获取结果
    result = first_completed_future.result()
    print(f"最快完成的任务结果是: {result}")

    # 此时，程序已经获取到结果并可以继续执行后续逻辑。
    # 耗时较长的任务 two 会在后台继续运行，直到完成，但我们已经不再关心它的结果。

    # 验证总耗时，这里应该接近 1 秒
    # 注意：如果在此处直接 time.sleep(5) 可能会等待任务 two 完成，
    # 但我们关注的是获取结果的时间点。
    # 为了演示，我们可以在任务函数中打印完成信息，以观察后台执行。

代码解析：

1. import concurrent.futures 和 import time: 导入所需模块。
2. def one() 和 def two(): 定义了两个模拟耗时任务。one耗时1秒，two耗时5秒。
3. with concurrent.futures.ThreadPoolExecutor() as pool:: 创建一个线程池执行器。使用with语句是最佳实践，它能确保在代码块结束后，线程池被正确关闭，释放资源。
4. tasks = [pool.submit(one), pool.submit(two)]: 将one和two函数提交给线程池。submit方法会立即返回两个Future对象，它们分别代表了one和two任务的未来结果。
5. first_completed_future = next(concurrent.futures.as_completed(tasks)): 这是核心步骤。
   • concurrent.futures.as_completed(tasks)会创建一个迭代器，它会在tasks列表中的任何一个Future对象完成时，立即将其 yield 出来。
   • next()函数用于从这个迭代器中获取第一个元素。因此，一旦one()或two()中的任意一个任务完成，as_completed就会产生对应的Future对象，next()就会捕获到它。由于one()仅耗时1秒，它会比two()先完成，所以first_completed_future将是one()任务对应的Future。
6. result = first_completed_future.result(): 调用Future对象的result()方法来获取任务的返回值。由于first_completed_future已经是完成状态，此调用将立即返回1。
7. 输出: 程序的总执行时间将主要由最快的任务决定（约1秒），并且输出结果为1。任务two会在后台继续执行，但其结果已被忽略。

优势与注意事项

• 简洁高效: concurrent.futures提供了一种声明式的方式来管理并发任务，大大减少了手动线程管理和同步的复杂性。
• 非阻塞获取: as_completed的特性使得我们能够以非阻塞的方式获取第一个完成的任务结果，而无需等待所有任务。
• 资源管理: ThreadPoolExecutor（通过with语句）自动处理线程的创建、复用和关闭，避免了资源泄露。
• 错误处理: Future.result()方法会重新抛出任务执行过程中发生的任何异常，使得错误处理更加直观。
• 适用场景: 这种模式非常适合需要“赛跑”的场景，例如向多个服务发送请求并只采纳第一个响应，或者在多个计算策略中选择最快的一个。
• 选择执行器:
  • ThreadPoolExecutor: 适用于I/O密集型任务（如网络请求、文件读写），因为Python的GIL（全局解释器锁）对CPU密集型任务的线程并发性能有较大限制。
  • ProcessPoolExecutor: 适用于CPU密集型任务，因为它使用独立的进程，每个进程有自己的Python解释器和内存空间，不受GIL限制。

总结

当需要在Python并发环境中启动多个任务，并快速获取其中最先完成的任务结果时，concurrent.futures模块提供了优雅且高效的解决方案。通过结合ThreadPoolExecutor进行任务提交和as_completed进行按完成顺序的迭代，我们可以轻松实现这一目标，从而优化程序的响应速度和资源利用率。这种模式是现代Python并发编程中处理“最快结果”场景的推荐方法。

以上就是Python并发编程：高效获取最快完成任务的结果的详细内容，更多请关注php中文网其它相关文章！