python-concurrent

概述

__all__ = (
    'FIRST_COMPLETED',
    'FIRST_EXCEPTION',
    'ALL_COMPLETED',
    'CancelledError',
    'TimeoutError',
    'BrokenExecutor',
    'Future',
    'Executor',
    'wait',
    'as_completed',
    'ProcessPoolExecutor',
    'ThreadPoolExecutor',
)

Executor

Executor 是一个抽象类，不应该直接使用此类，而是使用它提供的两个子类：ThreadPoolExecutor 和 ProcessPoolExecutor

Future

Futute 类封装可调用对象的异步执行，应由Executor.submit() 方法创建

concurrent.futures

异步并行任务编辑模块，提供一个高级的异步可执行的便利接口。

• concurrent.functures库统一了线程池和进程池

• ThreadPoolExecutor 异步调用的线程池的 Executor

• ProcessPoolExecutor 异步调用的进程池的 Executor

ThreadPoolExecutor线程池与ProcessPoolExecutor进程池对象

ThreadPoolExecutor

线程构造

ThreadPoolExecutor(max_workers=None, thread_name_prefix=’’, initializer=None, initargs=())->Future

max_workers 最大线程个数
mp_context 允许用户控制池创建的worker进程的start_method
initializer 每个工作进程创建开始时调用的函数，即进程创建后的第一个初始化函数
initargs 初始化 initializer 中的位置参数

from concurrent.futures import ThreadPoolExecutor
import time
import logging
import threading
import sys

FORMAT = "%(asctime)-15s [%(processName)s:%(threadName)s, %(process)d:%(thread)8d] %(message)s"
logging.basicConfig(stream=sys.stdout,level=logging.INFO,format=FORMAT)
logger = logging.getLogger(__name__)

def worker(n):
    logger.info(f'start thread {n}')
    time.sleep(n+1)
    logger.info(f'end thread {n}')
    return n

executor = ThreadPoolExecutor(max_workers=3)

futurelist = []
for i in range(3):
    logger.info(f"开始异步 {i}")
    future = executor.submit(worker, i)     # 异步-非阻塞-直接执行-执行完毕后立马执行下一条命令
    futurelist.append(future)

while True:
    logger.info(f"当前线程情况 {threading.enumerate()}")
    
    
    flag = True
    for i,f in enumerate(futurelist):
        logger.info(f"是否执行结束 {i} {f.done()}")
        flag = flag and f.done()

    if flag:
        executor.shutdown()  # 清理线程池
        logger.info(f"全部执行完毕 {threading.enumerate()}")
        break
    
    time.sleep(1)

for f in futurelist:
    logger.info(f.result())

ProcessPoolExecutor

进程构造

concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None, initializer=None, initargs=())

max_workers 最大进程个数
mp_context 允许用户控制池创建的worker进程的start_method
initializer 每个工作进程创建开始时调用的函数，即进程创建后的第一个初始化函数
initargs  初始化 initializer 中的位置参数

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from concurrent import futures
import threading,multiprocessing
import logging
import time,sys

FORMAT = "%(asctime)-15s [%(processName)s:%(threadName)s, %(process)d:%(thread)8d] %(message)s"
logging.basicConfig(stream=sys.stdout,level=logging.INFO,format=FORMAT)
logger = logging.getLogger(__name__)


def init():
    logger.info("初始化任务>>>>")

def worker(i):
    logger.info(f"开始工作了 {i}")
    time.sleep(i+1)
    logger.info(f"工作完成 {i}")
    return i

def callback(fs):
    logger.info(f"回调函数执行{fs.result()}")

def callback2(fs):
     logger.info(f"222回调函数执行{fs.result()}")


if __name__ == "__main__":
    executerPro = ProcessPoolExecutor(max_workers=3,  initializer=init)

    time.sleep(0.1)
    # 注意 ProcessPoolExecutor 是懒惰模式，使用 submit 才会创建  
    print(threading.enumerate(),multiprocessing.active_children())

    fs = []
    for i in range(3):
        f = executerPro.submit(worker,i)
        f.add_done_callback(callback)
        f.add_done_callback(callback2)
        fs.append(f)

    print(threading.enumerate())
    print(multiprocessing.active_children())
    

    done,not_done = futures.wait(fs,return_when=futures.ALL_COMPLETED)  # 当所有执行完毕后返回,见如下wait解析
    for i in done:
        print(i.result())
    print("开始打印not_down....")
    for i in not_done:
        print(i.result())

wait 和 as_completed

wait(fs, timeout=None, return_when=ALL_COMPLETED)

• 遍历future（可能由不同的Executor实例创建），并等待执行完成，如果未设置timeout参数，则代表不限制等待时间，return_when参数则用于设置次函数应该在何时返回，支持的选项如下：

按照线程池中线程完成的顺序返回， 而不是线程加载的顺序返回

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import random

def do_something(i):
    
    rad=random.randint(1, 5)
    print (f'线程{i}, 休眠{rad}秒')
    time.sleep(rad)
    return f'休眠完毕{i},{rad}'

start_time = time.perf_counter()

executor=ThreadPoolExecutor()
results = [executor.submit(do_something, i) for i in range(10)]
print(results)
for f in as_completed(results):
    print (f.result())

end_time = time.perf_counter()-start_time

print (f'总共耗时{round(end_time,2)}秒')

# 非指定顺序返回
休眠完毕2,1
休眠完毕8,1
休眠完毕6,2
休眠完毕1,2
休眠完毕4,3
休眠完毕5,4
休眠完毕9,4
休眠完毕7,5
休眠完毕3,5
休眠完毕0,5
总共耗时5.01秒

映射计算map

map方法则返回迭代器，在出现异常的时候，会全部报错，详细见异常处理

from concurrent.futures import ThreadPoolExecutor
import time

def do_something(seconds):
    print (f'休眠{seconds}秒')
    time.sleep(seconds)
    return '休眠完毕'

start_time = time.perf_counter()

executor=ThreadPoolExecutor()
sec = [5,4,3,2,1]
results = executor.map(do_something, sec)
for result in results:
 print (result)

end_time = time.perf_counter()-start_time

print (f'总共耗时{round(end_time,2)}秒')

支持上下文管理

使用with来避免使用 shutdown(wait=True)

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    future1 = executor.submit(f, 0)
    future2 = executor.submit(f, 0)
    future3 = executor.submit(f, 20)

异常处理

submit 方法返回的是 Future 对象， map方法则返回迭代器，如果没有调用future对象的result方法，即使执行过程中有异常用户也是不知道的

f = lambda x: 100 // x

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    future1 = executor.submit(f, 0)
    future2 = executor.submit(f, 0)
    future3 = executor.submit(f, 20)

todos = [future1, future2, future3]
for future in concurrent.futures.as_completed(todos):
    try:
        print(future.result())
    except ZeroDivisionError as e:
        print(e.__repr__())

>>>
ZeroDivisionError('integer division or modulo by zero')
5
ZeroDivisionError('integer division or modulo by zero')

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    # If a func call raises an exception, then that exception will be raised when its value is retrieved from the iterator.
    futures = executor.map(f, [0, 10, 20])

while True:
    try:
        future = futures.__next__()
    except StopIteration:
        break
    except ZeroDivisionError as e:
        print(e.__repr__())
 
>>>
ZeroDivisionError('integer division or modulo by zero')

参考文件

• Python-concurrent并发包
• https://zhuanlan.zhihu.com/p/343232126
• http://events.jianshu.io/p/0921638925f7