当前位置：首页 > 科技 > 软件

如何通过Python线程池实现异步编程？

来源：责编：时间：2023-08-05 11:45:38 3507观看

导读线程池的概念和基本原理线程池是一种并发处理机制，它可以在程序启动时创建一组线程，并将它们置于等待任务的状态。当任务到达时，线程池中的某个线程会被唤醒并执行任务，执行完任务后线程会返回线程池，等待下一个任务的到来

线程池的概念和基本原理

线程池是一种并发处理机制，它可以在程序启动时创建一组线程，并将它们置于等待任务的状态。当任务到达时，线程池中的某个线程会被唤醒并执行任务，执行完任务后线程会返回线程池，等待下一个任务的到来。这种机制可以减少线程的创建和销毁，提高程序的性能和效率。

线程池的基本原理是将任务和线程分离，将任务提交给线程池，由线程池来管理和执行任务。线程池中的线程可以被重复利用，减少了创建和销毁线程的开销，提高了程序的性能和效率。

Python 中线程池的实现方式

在 Python 中，线程池可以通过 concurrent.futures 模块中的 ThreadPoolExecutor 类来实现。这个类提供了一些方法来创建和管理线程池，以及提交和执行任务。

一、Python线程池的创建和销毁

创建线程池

在 Python 中，可以使用 concurrent.futures 模块中的 ThreadPoolExecutor 类来创建线程池。ThreadPoolExecutor 类的构造函数可以接受一个参数 max_workers，用于指定线程池的大小。如果不指定 max_workers，则线程池的大小会根据 CPU 的核心数来自动确定。

import concurrent.futuresdef task():    print('Task executed')if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        future = executor.submit(task)

在上述代码中，创建了一个包含三个线程的线程池，并提交了一个任务。使用 with 语句可以自动关闭线程池，确保资源的正确释放。

销毁线程池

要销毁线程池，可以调用 shutdown() 方法。该方法会等待所有任务执行完毕后再关闭线程池。

import concurrent.futuresdef task():    print('Task executed')if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        future = executor.submit(task)    executor.shutdown()

在上述代码中，关闭了线程池。

如果要立即关闭线程池，可以调用 shutdown(wait=False) 方法。该方法会立即关闭线程池，未完成的任务会被取消。这种方式需要特别小心，因为未完成的任务可能会导致程序的异常退出或数据丢失。

import concurrent.futuresdef task():    print('Task executed')if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        future = executor.submit(task)    executor.shutdown(wait=False)

在上述代码中，立即关闭了线程池。

线程池的生命周期

线程池的生命周期包括三个阶段：

创建阶段：创建线程池，并初始化线程池中的线程。
执行阶段：接收任务并执行任务，直到所有任务执行完毕或线程池被关闭。
销毁阶段：关闭线程池，释放所有资源。

在执行阶段中，无论是任务执行成功还是失败，都需要将线程返回线程池，以便线程池继续利用。如果任务执行失败，可以使用 Future 对象的 exception() 方法获取异常信息。

import concurrent.futuresdef task():    print('Task executed')    raise Exception('Task failed')if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        future = executor.submit(task)        try:            result = future.result()        except Exception as e:            print(f'Task failed: {e}')

在上述代码中，提交了一个会抛出异常的任务，并使用 try...except 语句来捕获异常信息。

在销毁阶段中，需要确保所有任务执行完毕后再关闭线程池。如果直接关闭线程池，未完成的任务可能会导致程序的异常退出或数据丢失。

线程池的异常处理

在使用线程池时，可能会出现各种异常，例如任务执行失败、线程池关闭失败等。为了保证程序的健壮性和可靠性，需要对这些异常进行处理。

在任务执行失败时，可以使用 Future 对象的 exception() 方法获取异常信息。在线程池关闭失败时，可以使用 ThreadPoolExecutor 类的 shutdown() 方法的返回值来判断是否成功关闭线程池。

import concurrent.futuresdef task():    print('Task executed')    raise Exception('Task failed')if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        future = executor.submit(task)        try:            result = future.result()        except Exception as e:            print(f'Task failed: {e}')        success = executor.shutdown(wait=False)        if not success:            print('Failed to shutdown thread pool')

在上述代码中，提交了一个会抛出异常的任务，并使用 try...except 语句来捕获异常信息。在关闭线程池时，使用 wait=False 参数来立即关闭线程池，并使用 shutdown() 方法的返回值来判断是否成功关闭线程池。

二、Python线程池的任务提交和执行

提交任务到线程池

要提交任务到线程池中，可以使用 submit() 方法，该方法会返回一个 Future 对象，表示任务的执行结果。

import concurrent.futuresdef task():    print('Task executed')    return 'Task result'if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        future = executor.submit(task)        print(future.result())

在上述代码中，提交了一个任务，并使用 future.result() 方法获取任务的执行结果。

可以使用 map() 方法来批量提交任务，并获得所有任务的执行结果。

import concurrent.futuresdef task(i):    print(f'Task {i} executed')    return f'Task {i} result'if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        results = executor.map(task, range(5))        for result in results:            print(result)

在上述代码中，使用 map() 方法批量提交任务，并获得所有任务的执行结果。

控制任务的执行顺序

在默认情况下，线程池会根据任务的提交顺序来执行任务。但是，如果需要控制任务的执行顺序，可以使用 submit() 方法的返回值 Future 对象来控制任务的执行。

import concurrent.futuresdef task(i):    print(f'Task {i} executed')    return f'Task {i} result'if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        futures = [executor.submit(task, i) for i in range(5)]        for future in concurrent.futures.as_completed(futures):            result = future.result()            print(result)

在上述代码中，使用 submit() 方法提交了多个任务，并将返回值 Future 对象保存在列表中。使用
concurrent.futures.as_completed() 函数来获取任务的执行结果，并按照完成顺序输出结果。

还可以使用 future.add_done_callback() 方法来注册回调函数，当任务执行完毕时自动调用回调函数。

import concurrent.futuresdef task(i):    print(f'Task {i} executed')    return f'Task {i} result'def callback(future):    result = future.result()    print(result)if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        futures = [executor.submit(task, i) for i in range(5)]        for future in futures:            future.add_done_callback(callback)

在上述代码中，使用 submit() 方法提交了多个任务，并使用 future.add_done_callback() 方法注册回调函数。当任务执行完毕时，会自动调用回调函数。

取消任务的执行

在使用线程池时，可能需要取消正在执行的任务。可以使用 Future 对象的 cancel() 方法来取消任务的执行。如果任务已经执行完毕或无法取消，cancel() 方法会返回 False。

import concurrent.futuresimport timedef task():    print('Task started')    time.sleep(5)    print('Task finished')    return 'Task result'if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        future = executor.submit(task)        time.sleep(2)        canceled = future.cancel()        if canceled:            print('Task canceled')        else:            print('Task not canceled')

在上述代码中，提交一个任务并等待 2 秒后取消任务的执行。如果任务已经执行完毕或无法取消，cancel() 方法会返回 False。

等待所有任务执行完毕

在使用线程池时，可能需要等待所有任务执行完毕。可以使用 wait() 方法来等待所有任务执行完毕。

import concurrent.futuresdef task(i):    print(f'Task {i} executed')    return f'Task {i} result'if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        futures = [executor.submit(task, i) for i in range(5)]        concurrent.futures.wait(futures)        for future in futures:            result = future.result()            print(result)

在上述代码中，使用 submit() 方法提交了多个任务，并将返回值 Future 对象保存在列表中。使用 concurrent.futures.wait() 函数来等待所有任务执行完毕。

三、Python线程池的参数和配置

下面是对 Python 中线程池的参数和配置的深入讲解。

线程池的大小

线程池的大小决定了可以同时执行的任务数。在 Python 中，可以使用 max_workers 参数来配置线程池的大小。如果不指定 max_workers，线程池的大小会根据 CPU 的核心数来自动确定。

import concurrent.futuresdef task():    print('Task executed')if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        future = executor.submit(task)

在上述代码中，创建了一个包含三个线程的线程池。如果需要更改线程池的大小，只需修改 max_workers 的值即可。

线程池的超时设置

在 Python 中，可以使用 timeout 参数来设置任务的执行超时时间。如果任务在指定的时间内没有执行完毕，线程池会自动取消任务的执行，并抛出 concurrent.futures.TimeoutError 异常。

import concurrent.futuresimport timedef task():    print('Task started')    time.sleep(5)    print('Task finished')    return 'Task result'if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:        future = executor.submit(task)        try:            result = future.result(timeout=2)            print(result)        except concurrent.futures.TimeoutError:            print('Task timeout')

在上述代码中，提交了一个需要 5 秒才能执行完毕的任务，并设置超时时间为 2 秒。因为任务没有在指定时间内执行完毕，所以会抛出 concurrent.futures.TimeoutError 异常。

线程池的任务队列

在线程池中，如果所有线程都正在执行任务，新的任务会被加入到任务队列中等待执行。在 Python 中，可以使用 queue_size 参数来配置任务队列的大小。如果任务队列已满，新的任务会被拒绝执行，并抛出 concurrent.futures.ThreadPoolExecutor 异常。

import concurrent.futuresdef task():    print('Task executed')if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3, queue_size=2) as executor:        for i in range(5):            future = executor.submit(task)

在上述代码中，创建了一个包含三个线程和大小为 2 的任务队列的线程池。提交了 5 个任务，其中前两个任务会被立即执行，后三个任务会被加入到任务队列中等待执行。因为任务队列只能容纳 2 个任务，所以第四个任务会被拒绝执行，并抛出 concurrent.futures.ThreadPoolExecutor 异常。

线程池的线程名称和优先级

在线程池中，可以为每个线程设置名称和优先级。在 Python 中，可以使用 thread_name_prefix 和 thread_priority 参数来配置线程名称和优先级。

import concurrent.futuresimport threadingdef task():    print(f'Task executed by {threading.current_thread().name}')if __name__ == '__main__':    with concurrent.futures.ThreadPoolExecutor(max_workers=3, thread_name_prefix='MyThread-', thread_priority=1) as executor:        future = executor.submit(task)

在上述代码中，创建了一个包含三个线程的线程池，并为每个线程设置名称前缀为 MyThread-，优先级为 1。提交了一个任务，任务会被其中一个线程执行，并在执行时输出线程的名称。

四、线程池的应用场景

线程池适用于需要并发执行多个任务的场景，例如：

网络爬虫：同时爬取多个网页。
数据库操作：同时查询多个数据表。
图像处理：同时处理多张图片。
并发编程：同时执行多个线程。

使用线程池可以减少线程的创建和销毁，提高程序的性能和效率，同时还可以控制线程池的大小和任务的执行顺序。

总之，线程池是一个非常有用的并发处理机制，可以提高程序的性能和效率，同时也需要仔细设计和实现，以避免并发问题和线程安全问题。

本文链接：http://www.28at.com/showinfo-26-124-0.html如何通过Python线程池实现异步编程？

声明：本网页内容旨在传播知识，若有侵权等问题请及时与本网联系，我们将在第一时间删除处理。邮件：2376512515@qq.com

上一篇：之家push系统迭代之路

下一篇：三言两语说透柯里化和反柯里化

标签：

热门焦点

5月iOS设备好评榜：iPhone 14仅排第43？

来到新的一月，安兔兔的各个榜单又重新汇总了数据，像安卓阵营的榜单都有着比较大的变动，不过iOS由于设备的更新换代并没有那么快，所以相对来说变化并不大，特别是iOS好评榜，老款设
一文看懂为苹果Vision Pro开发应用程序

译者 | 布加迪审校 | 重楼苹果的Vision Pro是一款混合现实（MR）头戴设备。Vision Pro结合了虚拟现实（VR）和增强现实（AR）的沉浸感。其高分辨率显示屏、先进的传感器和强大的处理能力
企业采用CRM系统的11个好处

客户关系管理（CRM）软件可以为企业提供很多的好处，从客户保留到提高生产力。　　CRM软件用于企业收集客户互动，以改善客户体验和满意度。　　CRM软件市场规模如今超过580
当家的盒马，加速谋生

来源 | 价值星球Planet作者 | 归去来自己“当家”的盒马，开始加速谋生了。据盒马官微消息，盒马计划今年开放生鲜供应链，将其生鲜商品送往食堂。目前，盒马在上海已经与
网传小米汽车开始筛选交付中心建筑面积不低于3000平方米

7月7日消息，近日有微博网友@长三角行健者爆料称，据经销商集团反馈，小米汽车目前已经开始了交付中心的筛选工作，要求候选场地至少有120个车位，建筑不能低
华为HarmonyOS 4.0将于8月4日发布或搭载AI大模型技术

华为宣布HarmonyOS4.0将于8月4日正式发布。此前，华为已经针对开发者公布了HarmonyOS4.0，以便于开发者提前进行适配，也因此被曝光出了一些新系统的特性
超级标准版旗舰！iQOO 11S全球首发iQOO超算独显芯片

上半年已接近尾声，截至目前各大品牌旗下的顶级旗舰都已悉数亮相，而下半年即将推出的顶级旗舰已经成为了数码圈爆料的主流，其中就包括全新的iQOO 11S系
最薄的14英寸游戏笔记本电脑 Alienware X14已可以购买

2022年1月份在国际消费电子展(CES2022)上首次亮相的Alienware新品——Alienware X14现在已经可以购买了，这款笔记本电脑被誉为世界上最薄的 14 英寸游戏笔
外交部：美方应停止在网络安全问题上不负责任地指责他国

　中国外交部今天（16日）举行例行记者会。会上，有记者问，美国情报官员称，他们正在阻拦来自中国以及其他国家的黑客获取相关科研成果。中方对此有何评论？对此