Python程序中的线程操作-concurrent模块

一、Python标准模块——concurrent.futures

官方文档：https://docs.python.org/dev/library/concurrent.futures.html

二、介绍

concurrent.futures模块提供了高度封装的异步调用接口

ThreadPoolExecutor：线程池，提供异步调用

ProcessPoolExecutor：进程池，提供异步调用

两者都实现相同的接口，该接口由抽象Executor类定义。

三、基本方法

submit(fn,*args,**kwargs)：异步提交任务

map(func,*iterables,timeout=None,chunksize=1)：取代for循环submit的操作

shutdown(wait=True)：相当于进程池的pool.close()+pool.join()操作

• wait=True，等待池内所有任务执行完毕回收完资源后才继续
• wait=False，立即返回，并不会等待池内的任务执行完毕
• 但不管wait参数为何值，整个程序都会等到所有任务执行完毕
• submit和map必须在shutdown之前

result(timeout=None)：取得结果

add_done_callback(fn)：回调函数

done()：判断某一个线程是否完成

cancle()：取消某个任务

四、ProcessPoolExecutor

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from threading import currentThread
from multiprocessing import current_process
import time,os


def task(i):
    # print(f'{currentThread().name} 在运行 任务{i}')
    print(f'{current_process().name} 在运行 任务{i}')
    time.sleep(0.2)
    return i**2

if __name__ == '__main__':
    pool = ProcessPoolExecutor(4)
    fu_list = []
    for i in range(20):
        future = pool.submit(task,i)
        # print(future.result())  # 拿不到值会阻塞在这里。
        fu_list.append(future)

    pool.shutdown(wait=True)  # 等待池内所有任务执行完毕
    for i in fu_list:
        print(i.result())# 拿不到值会阻塞在这里。

五、ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor,os


def task(i):
    # print(f'{currentThread().name} 在运行 任务{i}')
    print(f'{current_process().name} 在运行 任务{i}')
    time.sleep(0.2)
    return i**2

if __name__ == '__main__':
    pool = ThreadPoolExecutor(4)
    fu_list = []
    for i in range(20):
        future = pool.submit(task,i)
        # print(future.result())  # 拿不到值会阻塞在这里。
        fu_list.append(future)

    pool.shutdown(wait=True)  # 等待池内所有任务执行完毕
    for i in fu_list:
        print(i.result())# 拿不到值会阻塞在这里。

六、回调函数

from concurrent.futures import ThreadPoolExecutor,os


def task(i):
    # print(f'{currentThread().name} 在运行 任务{i}')
    print(f'{current_process().name} 在运行 任务{i}')
    time.sleep(0.2)
    return i**2
def parse(future):
    # print(future.result())
    # print(currentThread().name,'拿到了结果',future.result()) # 如果是线程池 执行完当前任务 负责执行回调函数的是执行任务的线程。
    print(current_process().name,future.result()) # 如果是进程池 执行完当前任务 负责执行回调函数的是执行任务的是主进程


if __name__ == '__main__':
    pool = ProcessPoolExecutor(4)
    # pool = ThreadPoolExecutor(4)
    fu_list = []
    for i in range(20):
        future = pool.submit(task,i)
        future.add_done_callback(parse) # 绑定回调函数
        # 当任务执行结束拿到返回值的时候自动触发回调函数。并且把future当做参数直接传给回调函数parse