Python concurrent.futures 线程池进程池模块

在python3.2版本及以上使用concurrent.futures进行线程池和进程池的使用，实现数据高并发。

如果一个人心里藏了那么多的喜怒哀乐，一定曾经活得伤痕累累。

在Python3.2开始就有concurrent_futures，其可以实现线程池，进程池，不必再自己使用管道传数据造成死锁的问题。并且这个模块具有线程池和进程池、管理并行编程任务、处理非确定性的执行流程、进程/线程同步等功能，但是平时用的最多的还是用来构建线程池和进程池。

此模块由以下部分组成：

concurrent.futures.Executor: 这是一个虚拟基类，提供了异步执行的方法。
submit(function, argument): 调度函数（可调用的对象）的执行，将 argument 作为参数传入。
map(function, argument): 将 argument 作为参数执行函数，以异步的方式。
shutdown(Wait=True): 发出让执行者释放所有资源的信号。
concurrent.futures.Future: 其中包括函数的异步执行。Future对象是submit任务（即带有参数的functions）到executor的实例

知识回顾

线程与线程池
 进程与进程池
 异步协程

线程池案例

# coding:utf-8
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
# 导入线程池和进程池模块
import threading
# 导入线程模块，作用是获取当前线程的名称
import os,time

def task(n):
    print('%s:%s is running' %(threading.currentThread().getName(),os.getpid()))
    # 打印当前线程名和运行的id号码
    time.sleep(2)
    return n**2
    # 返回传入参数的二次幂

if __name__ == '__main__':
    p=ThreadPoolExecutor()
    #实例化线程池，设置线程池的数量，不填则默认为cpu的个数*5
    l=[]
    # 用来保存返回的数据，做计算总计
    for i in range(10):
        obj=p.submit(task,i)
        # 传入的参数是要执行的函数和该函数接受的参数
        # -----------------------------------
        # 这里执行的方式是异步执行
        # # p.submit(task,i).result()即同步执行
        # -----------------------------------
        # 上面的方法使用range循环有个高级的写法，即map内置函数
        # obj=p.map(task,range(10))
        # p.shutdown()
        # -----------------------------------
        l.append(obj)
        # 把返回的结果保存在空的列表中，做总计算
    p.shutdown()
    # 所有计划运行完毕，关闭结束线程池

    print('='*30)
    print([obj.result() for obj in l])

#上面方法也可写成下面的方法
    # with ThreadPoolExecutor() as p:   #类似打开文件,可省去.shutdown()
    #     future_tasks = [p.submit(task, i) for i in range(10)]
    # print('=' * 30)
    # print([obj.result() for obj in future_tasks])

进程池案例

from concurrent.futures import ProcessPoolExecutor
import os,time,random
def task(n):
    print('%s is running' %os.getpid())
    time.sleep(2)
    return n**2


if __name__ == '__main__':
    p=ProcessPoolExecutor()  #不填则默认为cpu的个数
    l=[]
    start=time.time()
    for i in range(10):
        obj=p.submit(task,i)   #submit()方法返回的是一个future实例，要得到结果需要用obj.result()
        l.append(obj)

    p.shutdown()  #类似用from multiprocessing import Pool实现进程池中的close及join一起的作用
    print('='*30)
    # print([obj for obj in l])
    print([obj.result() for obj in l])
    print(time.time()-start)

    #上面方法也可写成下面的方法
    # start = time.time()
    # with ProcessPoolExecutor() as p:   #类似打开文件,可省去.shutdown()
    #     future_tasks = [p.submit(task, i) for i in range(10)]
    # print('=' * 30)
    # print([obj.result() for obj in future_tasks])
    # print(time.time() - start)

refs

进程池、线程池、回调函数 - 听风。 - 博客园
 使用concurrent.futures模块并发，实现进程池、线程池 - 听风。 - 博客园
 Python concurrent.futures 线程池进程池模块 | Langzi - Never Setter 永不将就
 Python concurrent.futures 线程池进程池模块 | Langzi - Never Setter 永不将就