使用Future、asyncio处理并发

并发的意义

为了高效处理网络I/O，需要使用并发，因为网络有很高的延迟，所以为了不浪费CPU周期去等待，最好在收到网络响应之前做些其他的事。

在I/O密集型应用中，如果代码写得正确，那么不管是用哪种并发策略（使用线程或asyncio包），吞吐量都比依序执行的代码高很多。

　　并发是指一次处理多件事。并行是指一次做多件事。一个关于结构，一个关于执行。

并行才是我们通常认为的那个同时做多件事情，而并发则是在线程这个模型下产生的概念。

并发表示同时发生了多件事情，通过时间片切换，哪怕只有单一的核心，也可以实现“同时做多件事情”这个效果。

根据底层是否有多处理器，并发与并行是可以等效的，这并不是两个互斥的概念。

举个我们开发中会遇到的例子，我们说资源请求并发数达到了1万。这里的意思是有1万个请求同时过来了。但是这里很明显不可能真正的同时去处理这1万个请求的吧！

如果这台机器的处理器有4个核心，不考虑超线程，那么我们认为同时会有4个线程在跑。

也就是说，并发访问数是1万，而底层真实的并行处理的请求数是4。

如果并发数小一些只有4的话，又或者你的机器牛逼有1万个核心，那并发在这里和并行一个效果。

也就是说，并发可以是虚拟的同时执行，也可以是真的同时执行。而并行的意思是真的同时执行。

结论是：并行是我们物理时空观下的同时执行，而并发则是操作系统用线程这个模型抽象之后站在线程的视角上看到的“同时”执行。

Future

一、初识future

concurrent.futures 模块主要特色是：ThreadPoolEXecutor和 ProcessPoolExecutor类，这两个类实现的接口能分别在不同的线程或进程中执行可调用的对象。

这两个类在内部维护着一个工作线程或进程池，以及要执行的任务队列。

from concurrent import futures

MAX_WORKERS = 20

def download_many():

    workers = min(MAX_WORKERS,len(url_list))
    with futures.ThreadPoolExecutor(workers) as executor:
        res = executor.map(download_one,sorted(url_list))
    return len(list(res))

（1）设定工作的线程数量，使用允许的最大值与要处理的数量之间的较小的那个值，以免创建过于的线程。

（2）download_one函数在多个线程中并发调用，map方法返回一个生成器，因此可以迭代，获取各个函数返回的值。

future是concurrent.futures模块和asyncio包的重要组件。

从python3.4开始标准库中有两个名为Future的类：concurrent.futures.Future和asyncio.Future

这两个类的作用相同：两个Future类的实例都表示可能完成或者尚未完成的延迟计算。与Twisted中的Deferred类、Tornado框架中的Future类的功能类似

future封装待完成的操作，可以放入队列，完成的状态可以查询，得到结果（或抛出异常）后可以获取结果（或异常）。

▲ 通常情况下自己不应该创建future，只能由并发框架（concurrent.future或asyncio）实例化。

future表示终将发生的事情，而确定某件事会发生的唯一方式就是执行的时间已经排定。

只有排定把某件事交给concurrent.futures.Executor子类处理时，才会创建concurrent.futures.Future实例。

Executor.submit(fn, *args, **kwargs)

Executor.submit() 方法的参数是一个可调用的对象，调用这个方法后会为传入的可调用对象排期，返回一个future。

▲ 不是阻塞的，而是立即返回。能够使用 done()方法判断该任务是否结束。

使用cancel()方法可以取消提交的任务，如果任务已经在线程池中运行了，就取消不了。

客户端代码不应该改变future的状态，并发框架在future表示的延迟计算结束后会改变future状态。而我们无法控制计算何时结束。

Executor.shutdown(wait=True)

释放系统资源,在Executor.submit()或 Executor.map()等异步操作后调用。使用with语句可以避免显式调用此方法。

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

wait=True，等待池内所有任务执行完毕回收完资源后才继续，--------》默认

wait=False，立即返回，并不会等待池内的任务执行完毕

但不管wait参数为何值，整个程序都会等到所有任务执行完毕

Executor.add_done_callback(fn)

future都有 .add_done_callback(fn) 方法，这个方法只有一个参数，类型是可调用的对象，future运行结束后会调用指定的可调用对象。

fn接收一个future参数，通过obj.result()，获得执行后结果。

Executor.result()

.result()方法，在future运行结束后调用的话，返回可调用对象的结果，或者重新抛出执行可调用的对象时抛出的异常。

如果没有运行结束，concurrent会阻塞调用方直到有结果可返回。

concurrent.futures.as_completed()

使用concurrent.futures.as_completed函数，这个函数的参数是一个future列表 / future为key的字典，返回值是一个生成器，

在没有任务完成的时候，会阻塞，在有某个任务完成的时候，会yield这个任务future，就能执行for循环下面的语句，然后继续阻塞，循环到所有的任务结束。

从结果也可以看出，先完成的任务会先通知主线程。

Executor.map(func, *iterables, timeout=None) 

Executor.map() 返回值是一个迭代器，迭代器的__next__方法调用各个future的result()方法，得到各个future的结果而不是future本身。

*iterables：可迭代对象，如列表等。每一次func执行，都会从iterables中取参数。

timeout：设置每次异步操作的超时时间

修改Executor.map调用，换成两个for循环，一个用于创建并排定future，另一个用于获取future的结果

def download_many():

    with futures.ThreadPoolExecutor(max_workers=3) as executor:

        to_do = []
        for cc in sorted(url_list):
            future = executor.submit(download_one,cc)
            to_do.append(future)

        result = []
        for future in futures.as_completed(to_do):
            res = future.result()
            result.append(res)

executor.submit() 方法排定可调用对象的执行时间，然后返回一个future，表示这个待执行的操作。

示例中的future.result()方法绝不会阻塞，因为future由as_completed函数产出。

▲ 同时在 future.result()处使用 try模块捕获异常

二、阻塞型I/O和GIL

Cpython解释器本身就不是线程安全的，因此有全局解释器锁（GIL），一次只允许使用一个线程执行Python字节码。因此，一个Python进程通常不能同时使用多个CPU核心。

标准库中所有执行阻塞型I/O操作的函数，在等待操作系统返回结果时都会释放GIL。I/O密集型Python程序能从中受益。

一个Python线程等待网络响应时，阻塞型I/O函数会释放GIL，再运行一个线程。

三、ProcessPoolExecutor

　　ProcessPoolExecutor 和 ThreadPoolExecutor类都实现了通用的Executor接口，因此使用concurrent.futures模块能特别轻松地把基于线程的方案转成基于进程的方案。

　　ThreadPoolExecutor.__init__方法需要max_workers参数，指定线程池中线程的数量。（10、100或1000个线程）

　　ProcessPoolExecutor类中这个参数是可选的，而且大多数情况下不使用，默认值是os.cpu_count()函数返回的CPU数量。四核CPU，因此限制只能有4个并发。而线程池版本可以有上百个。

　　ProcessPoolExecutor类把工作分配给多个Python进程处理，因此，如果需要做CPU密集型处理，使用这个模块能绕开GIL，利用所有的CPU核心。

　　其原理是一个ProcessPoolExecutor创建了N个独立的Python解释器，N是系统上面可用的CPU核数。

　　使用方法和ThreadPoolExecutor方法一样

from time import sleep,strftime
from concurrent import futures

def display(*args):
    print(strftime('[%H:%M:%S]'),end=' ')
    print(*args)

def loiter(n):
    msg = '{}loiter({}): doing nothing for {}s'
    display(msg.format('	'*n,n,n))
    sleep(n*2)
    msg = '{}loiter({}): done.'
    display(msg.format('	'*n,n))
    return n *10

def main():
    display('Script starting...')
    executor = futures.ThreadPoolExecutor(max_workers=3)
    results = executor.map(loiter,range(5))
    display('result：',results)
    display('Waiting for individual results:')

    for i,result in enumerate(results):
        display('result {}:{}'.format(i,result))

main()

Executor.map函数返回结果的顺序与调用时开始的顺序一致。

如果第一个调用生成结果用时10秒，而其他调用只用1秒，代码会阻塞10秒，获取map方法返回的生成器产出的第一个结果。

在此之后，获取后续结果不会阻塞，因为后续的调用已经结束。

如果需要不管提交的顺序，只要有结果就获取，使用 Executor.submit() 和 Executor.as_completed() 函数。

四、显示下载进度条

　　TQDM包特别易于使用。

from tqdm import tqdm
import time

for i in tqdm(range(1000)):
    time.sleep(.01)

　　tqdm函数能处理任何可迭代的对象，生成一个迭代器。

　　使用这个迭代器时，显示进度条和完成全部迭代预计的剩余时间。

　　为了计算剩余时间，tqdm函数要获取一个能使用len函数确定大小的可迭代对象，或者在第二个参数中指定预期的元素数量。

　　如：iterable = tqdm.tqdm(iterable, total=len(xx_list))

Asyncio

一、使用asyncio包处理并发

　　这个包主要使用事件循环的协程实现并发。

import asyncio
import itertools
import sys

@asyncio.coroutine
def spin(msg):

    write,flush = sys.stdout.write,sys.stdout.flush

    for char in itertools.cycle('|/-\'):
        status = char + ' ' +msg
        write(status)
        flush()
        write('x08'*len(status))
        try:
            yield from asyncio.sleep(.1)
        except asyncio.CancelledError:
            break
    write(' '*len(status) + 'x08'*len(status))

@asyncio.coroutine
def slow_function():
    yield from asyncio.sleep(3)
    return 42

@asyncio.coroutine
def supervisor():
    spinner = asyncio.async(spin('thinking'))
    print('spinner object:',spinner)
    result = yield from slow_function()
    spinner.cancel()
    return result

def main():
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(supervisor())
    loop.close()
    print('Answer:',result)

　　（1）打算交给asyncio处理的协程要使用@asyncio.coroutine装饰。

　　（2）使用yield from asyncio.sleep 代替 time.sleep，这样休眠不会阻塞事件循环。

　　（3）asyncio.async(...)函数排定spin协程的运行时间，使用一个Task对象包装spin协程，并立即返回。

　　（4）获取事件循环的引用，驱动supervisor协程。

▲ 如果写成需要在一段时间内什么也不做，应该使用yield from asyncio.sleep(DELAY)

asyncio.Task对象差不多与threading.Thread对象等效，Task对象像是实现协作式多任务的库（如：gevent）中的绿色线程（green thread）

获取的Task对象已经排定了运行时间，Thread实例必须调用start方法，明确告知让他运行。

没有API能从外部终止线程，因为线程随时可能被中断，导致系统处于无效状态。

如果想要终止任务，使用Task.cancel()实例方法，抛出CancelledError异常。协程可以在暂停的yield处捕获这个异常，处理终止请求。

二、asyncio.Future 与 concurrent.futures.Future

　　asyncio.Future 与 concurrent.futures.Future类的接口基本一致，不过实现方式不同，不可以互换。

　　future只是调度执行某物的结果。

　　在asyncio包中，BaseEventLoop.create_task(...)方法接收一个协程，排定它的运行时间，然后返回一个asyncio.Task实例，也是asyncio.Future类的实例，因为Task是Future的子类，用于包装协程。

　　asyncio.Future类的目的是与yield from一起使用，所以通常不需要使用以下方法。

　　（1）无需调用my_future.add_done_callback(...)，因为可以直接把想在future运行结束后执行的操作放在协程中yield from my_future表达式的后面，

　　（2）无需调用my_future.result()，因为yield from从future中产出的值就是结果（result = yield from my_future）。

　　asyncio.Future对象由yield from驱动，而不是靠调用这些方法驱动。

　　获取Task对象有两种方式：

　　（1）asyncio.async(coro_or_future, *, loop=None)，

　　　　第一个参数如果是Future或者Task对象，返回。如果是协程，那么async函数会调用loop.create_task(...)方法创建Task对象。

　　（2）BaseEventLoop.create_task(coro)，

　　　　排定协程的执行时间，返回一个asyncio.Task对象。

三、asyncio和aiohttp

　　asyncio包只直接支持TCP和UDP。如果想使用HTTP或其他协议，那么要借助第三方包。

import asyncio
import aiohttp

@asyncio.coroutine
def get_flag(url):
    resp = yield from aiohttp.request('GET',url)
    data = yield from resp.read()
    return data

@asyncio.coroutine
def download_one(url):
    data = yield from get_flag(url)
    return url

def download_many():
    loop = asyncio.get_event_loop()
    to_do = [download_one(url) for url in sorted(url_list)]
    wait_coro = asyncio.wait(to_do)
    res,_ = loop.run_until_complete(wait_coro)
    loop.close()
    
    return len(res)

阻塞的操作通过协程实现，客户代码通过yield from把职责委托给协程，以便异步运行协程。

构建协程对象列表。

asyncio.wait是一个协程，等传给它的所有协程运行完毕后结束。wait函数默认行为。

loop.run_until_complete(wait_coro)执行事件循环。直到wait_coro运行结束；时间循环运行的过程中，这个脚本会在这里阻塞。

asyncio.wait函数运行结束后返回一个元组，第一个元素是一系列结束的future，第二个元素是一系列未结束的future。

（如果设置了timeout和return_when 就会返回未结束的future）

▲ 为了使用asyncio包，必须把每个访问网络的函数改成异步版，使用yield from处理网络操作，这样才能把控制权交还给事件循环。

总结：

　　（1）我们编写的协程链条始终是通过把最外层委派生成器传给asyncio包API中的某个函数（如loop.run_until_complete(...)）驱动。

　　　　由asyncio包实现next(...)或.send(...)

　　（2）我们编写的协程链条始终通过yield from把职责委托给asyncio包中的某个协程函数或协程方法（yield from asyncio.sleep(...)），或者其他库中实现高层协议的协程（yield from aiohttp.request(...)），

　　　　也就是说最内层的子生成器是库中真正执行I/O操作的函数，而不是我们自己编写的函数。

四、asyncio与进度条结合

　　由loop.run_until_complete方法驱动，全部协程运行完毕后，这个函数会返回所有下载结果。

　　可是，为了更新进度条，各个协程运行结束后就要立即获取结果。

import asyncio
import aiohttp
from tqdm import tqdm
import collections

@asyncio.coroutine
def get_flag(url):
    resp = yield from aiohttp.request('GET',url)
    data = yield from resp.read()
    return data

@asyncio.coroutine
def download_one(url,semaphore):

    try:
        with (yield from semaphore):
            data = yield from get_flag(url)
    except Exception as exc:
        ''''''
    else:
        save_data(data)
    return url

@asyncio.coroutine
def download_coro(url_list,concur_req):

    counter = collections.Counter()
    semaphore = asyncio.Semaphore(concur_req)

    to_do = [download_one(url,semaphore) for url in url_list]
    to_do_iter = asyncio.as_completed(to_do)

    to_do_iter =  tqdm(to_do_iter,total=len(url_list))
    for future in to_do_iter:

        try:
            res = yield from future
        except Exception as exc:
            ''''''
        counter[status] += 1
    return counter

def download_many():
    loop = asyncio.get_event_loop()
    coro = download_coro(url_list,concur_req)
    res = loop.run_until_complete(coro)
    loop.close()

    return res    

　　（1）使用某种限流机制，防止向服务器发起太多并发请求，使用ThreadPoolExecutor类时可以通过设置线程池数量；

　　（2）asyncio.Semaphore对象维护这一个内部计数器，把semaphore当做上下文管理器使用。保证任何时候都不会有超过X个协程启动。

　　（3）asyncio.as_completed(xxx)，获取一个迭代器，这个迭代器会在future运行结束后返回future。

　　（4）迭代运行结束的future，获取asyncio.Future对象的结果，使用yield from，而不是future.result()方法。

　　（5）不能使用字典映射方式，因为asyncio.as_completed函数返回的future与传给as_completed函数的future可能不同。在asyncio包内部，我们提供的future会被替换成生成相同结果的future。

五、使用Executor对象，防止阻塞事件循环

　　上述示例中，save_data(...)，会执行硬盘I/O操作，而这应该异步执行。

　　在线程版本中，save_data(...)会阻塞download_one函数的线程，但是阻塞的只是众多工作线程中的一个。

　　阻塞型I/O调用在背后会释放GIL，因此另一个线程可以继续。

　　但是在asyncio中，save_data(...)函数阻塞了客户代码与asyncio事件循环共用的唯一线程，因此保存文件时，整个应用程序都会冻结。

　　

　　asyncio的事件循环在背后维护者一个ThreadPoolExecutor对象，我们可以调用run_in_executor方法，把可调用对象发给它执行。

@asyncio.coroutine
def download_one(url,semaphore):

    try:
        with (yield from semaphore):
            data = yield from get_flag(url)
    except Exception as exc:
        ''''''
    else:
        loop = asyncio.get_event_loop()
        loop.run_in_executor(None, save_data, data)
    return url

　　（1）获取事件循环对象的引用。

　　（2）run_in_executor方法的第一个参数是Executor实例；如果设为None，使用事件循环的默认ThreadPoolExecutor实例。

　　（3）余下参数是可调用的对象，以及可调用对象的位置参数。

每次下载发起多次请求：

@asyncio.coroutine
def get_flag(url):
    resp = yield from aiohttp.request('GET',url)
    data = yield from resp.read()
    json = yield from resp.json()
    return data

@asyncio.coroutine
def download_one(url,semaphore):

    try:
        with (yield from semaphore):
            flag = yield from get_flag(url)
        with (yield from semaphore):
            country = yield from get_country(url)
    except Exception as exc:
        ''''''
    return url

六、使用asyncio包编写服务器