Python并行编程(十四)：异步编程

1、基本概念

　　除了顺序执行和并行执行的模型以外，还有异步模型，这是事件驱动模型的基础。异步活动的执行模型可以只有一个单一的主控制流，能在单核心系统和多核心系统中运行。

　　在并发执行的异步模型中，许多任务被穿插在同一时间线上，所有的任务都由一个控制流执行(单一线程)。任务的执行可能被暂停或恢复，中间的这段时间线程将会执行其他任务。大致如下：

　　

　　如上图所示，任务(不同的颜色表示不同的任务)可能被其他任务插入，但是都处在同一个线程下。这表明当某一个任务执行的时候，其他任务都暂停了。与多线程编程模型很大的一点不同是，多线程的某个任务在时间线上什么时候挂起某个活动或恢复某个活动由系统决定，而在异步中，程序员必须假设线程可能在任何时间被挂起和替换。

　　程序员可以将任务编写成许多可以间隔执行的小步骤，如果一个任务需要另一个任务的输出，那么被依赖的任务必须接收它的输入。

2、使用Python的concurrent.futures模块

　　这个模块具有线程池和进程池、管理并行编程任务、处理非确定性的执行流程、进程/线程同步等功能。

　　此模块由一下部分组成：

　　- concurrent.futures.Executor：这是一个虚拟基类，提供了异步执行的方法。

　　- submit(function, argument)：调度函数(可调用的对象)的执行，将argument作为参数传入。

　　- map(function, argument)：将argument作为参数执行函数，以异步的方式。

　　- shutdown(Wait=True)：发出让执行者释放所有资源的信号。

　　- concurrent.futures.Future：其中包括函数的异步执行。Future对象是submit任务(即带有参数的functions)到executor的实例。

　　Executor是抽象类，可以通过子类访问，即线程或进程的ExecutorPools。因为线程或进程的实例是依赖于资源的任务，所以最好以池的形式将他们组织在一起，作为可以重用的launcher和executor。

　　线程池和进程池是用于在程序中优化和简化线程/进程的使用。通过池可以提交任务给executor。池由两部分组成，一部分是内部的队列，存放着待执行的任务；另一部分是一系列的进程或线程，用于执行这些任务。池的概念主要目的是为了重用：让线程或进程在生命周期内可以多次使用。他减少了创建线程和进程的开销，提高了程序性能。重用不是必须的规则，但它是程序员在应用中使用池的主要原因。

　　

　　current.Futures提供了两种Executor的子类，各自独立操作一个线程池和一个进程池。这两个子类分别是：

　　- concurrent.futures.ThreadPoolExecutor(max_workers)

　　- concurrent.futures.ProcessPoolExecutor(max_workers)

　　max_workers参数表示最多有多少个worker并行执行任务

　　代码测试：

import concurrent.futures
import time

number_list = [1,2,3,4,5,6,7,8,9,10]

def evaluate_item(x):
    #For time consuming
    result_item = count(x)
    return result_item

def count(number):
    for i in range(0, 10000000):
        i = i + 1
    return i * number

if __name__ == "__main__":
    # Sequential execution
    start_time = time.time()
    for item in number_list:
        print(evaluate_item(item))
    print("Sequential execution in %s seconds" %(str(time.time() - start_time)))
    # Thread pool execution
    start_time_1 = time.time()
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(evaluate_item, item) for item in number_list]
        for future in concurrent.futures.as_completed(futures):
            print(future.result())
    print("Thread pool execution in %s seconds"  %(str(time.time() - start_time_1)))

    # Process pool execution
    start_time_2 = time.time()
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(evaluate_item, item) for item in number_list]
    print("Process pool execution in %s seconds" %(str(time.time() - start_time_2)))

　　运行结果：

10000000
20000000
30000000
40000000
50000000
60000000
70000000
80000000
90000000
100000000
Sequential execution in 8.975373029708862 seconds
10000000
20000000
30000000
40000000
60000000
70000000
50000000
80000000
90000000
100000000
Thread pool execution in 8.699156045913696 seconds
Process pool execution in 5.916198968887329 seconds

　　创建一个list存放10个数字，然后使用一个循环计算从1加到10000000，打印出和与number_list的乘积。

number_list = [1,2,3,4,5,6,7,8,9,10]

def evaluate_item(x):
    #For time consuming
    result_item = count(x)
    return result_item

def count(number):
    for i in range(0, 10000000):
        i = i + 1
    return i * number

　　在主程序中，首先顺序执行了一次程序并打印其执行时间：

start_time = time.time()
    for item in number_list:
        print(evaluate_item(item))
    print("Sequential execution in %s seconds" %(str(time.time() - start_time)))

　　其次使用futures.ThreadPoolExecutor模块的线程池并打印其时间：

start_time_1 = time.time()
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(evaluate_item, item) for item in number_list]
        for future in concurrent.futures.as_completed(futures):
            print(future.result())
    print("Thread pool execution in %s seconds"  %(str(time.time() - start_time_1)))

　　ThreadPoolExecutor使用线程池中的一个线程执行给定任务。池中一共有5个线程，每一个线程从池中取得一个任务然后执行它，当任务执行完成，再从池中拿到另一个任务。

　　最后是使用进程池：

start_time_2 = time.time()
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(evaluate_item, item) for item in number_list]
    print("Process pool execution in %s seconds" %(str(time.time() - start_time_2)))

　　和ThreadPoolExecutor一样，ProcessPoolExecutor是一个executor，使用一个线程池来并行执行任务。因为ProcessPoolExecutor使用了多核处理的模块，让我们可以不受GIL的限制，大大缩短执行时间。

　　几乎所有需要处理多个客户端请求的服务应用都会使用池。也有应用要求需要立即执行，或者要求对任务的线程有更多的控制器，这种情况下，池不是一个最佳选择。

3、使用Asyncio管理事件循环

　　先入为主：

import asyncio
import datetime
import time

def function_1(end_time, loop):
    print("function_1 called")
    if (loop.time() + 1.0) < end_time:
        loop.call_later(1, function_2, end_time, loop)
    else:
        loop.stop()

def function_2(end_time, loop):
    print("function_2 called")
    if (loop.time() + 1.0) < end_time:
        loop.call_later(1, function_3, end_time, loop)
    else:
        loop.stop()

def function_3(end_time, loop):
    print("function_3 called")
    if (loop.time() + 1.0) < end_time:
        loop.call_later(1, function_1, end_time, loop)
    else:
        loop.stop()

def function_4(end_time, loop):
    print("function_4 called")
    if (loop.time() + 1.0) < end_time:
        loop.call_later(1, function_4, end_time, loop)
    else:
        loop.stop()

loop = asyncio.get_event_loop()
print(loop.time())
end_loop = loop.time() + 9.0
print(end_loop)
loop.call_soon(function_1, end_loop, loop)
#loop.call_soon(function_4, end_loop, loop)
loop.run_forever()
loop.close()

　　执行结果：

　　

　　上述例子定义了三个异步任务，相继执行，如图所示：

　　

　　首先，我们要得到这个事件循环：

loop = asyncio.get_event_loop()

　　然后我们通过call_soon方法调用了function_1()函数。

end_loop = loop.time() + 9.0
loop.call_soon(function_1, end_loop, loop)

　　function_1：

def function_1(end_time, loop):
    print("function_1 called")
    if (loop.time() + 1.0) < end_time:
        loop.call_later(1, function_2, end_time, loop)
    else:
        loop.stop()

　　- end_time定义了function_1可以运行的最长时间，并通过call_later方法传入到function_2中作为参数

　　- loop通过get_event_loop()方法得到的事件循环

　　任务执行结束之后，它会比较loop.time() + 1s和设定的运行时间，如果没有超过，使用call_later在1秒之后执行function_2()，function_2和3作用类似

　　如果运行时间超过了设定，时间循环终止。

　　概念解释：

　　Python的Asyncio模块提供了管理事件、协程、任务和线程方法，以及编写并发代码的原语。主要组件和概念包括：

　　- 事件循环：在Asyncio模块中，每一个进程都有一个事件循环。

　　- 协程：这是子程序的泛化概念。协程可以在执行期间暂停，这样就可以等待外部的处理(例如IO)完成之后，从之前暂停的地方恢复执行。

　　- Futures：定义了Future对象，和concurrent.futures模块一样，表示尚未完成的计算。

　　- Tasks：这是Asyncio的子类，用于封装和管理并行模式下的协程。

　　事件循环：

　　　　在计算机系统中，可以产生事件的实体叫做事件源，能处理事件的实体叫做事件处理者，还有一些第三方实体叫做事件循环。它的作用是管理所有的事件，在整个程序运行过程中不断循环执行，追踪事件发生的顺序将他们放到队列中，当主线程空闲的时候，调用相应的事件处理者处理事件。

 　　Asyncio管理事件循环的方法：

　　- loop = get_event_loop()：得到当前上下文的事件循环。

　　- loop.call_later(time_delay, callback, argument)：延后time_delay秒再执行callback方法。

　　- loop.call_soon(callback, argument)：尽可能快的调用callback。call_soon()函数结束，主线程回到事件循环之后就会马上调用callback。

　　- loop.time()：以float类型返回当前时间循环的内部时间。

　　- asyncio.set_event_loop()：为当前上下文设置时间循环。

　　- asyncio.new_event_loop()：根据此策略创建一个新的时间循环并返回。

　　- loop.run_forever()：在调用stop()之前将一直运行。run_forever真正开始执行函数。

4、使用Asyncio管理协程

　　上述例子中一个程序变得很大而且复杂时，将其划分为子程序，每一部分实现特定的任务。子程序不能单独执行，只能在主程序的请求下执行，主程序负责协调使用各个子程序。协程是子程序的泛化，和子程序一样的是，协程只负责计算任务的一步；不同的是协程没有主程序来进行调度。因为协程通过管道连接在一起，没有监视函数负责顺序调用他们。在协程中，执行点可以被挂起，可以被之前挂起的点恢复执行。通过协程池就可以插入到计算中：运行第一个任务，直到它返回yield执行权，然后运行下一个，这样顺着执行下去。

　　这种插入的控制组件就是前文提到的事件循环，它持续追踪所有的协程并执行它们。

　　协程的另外一些重要特性如下：

　　- 协程可以有多个入口点，并可以yield多次

　　- 协程可以将执行权交给其他协程

　　yield表示协程在此暂停，并且将执行权交给其他协程，因为协程可以将值与控制权一起传递给另一个协程，所以yield一个值就表示将值传给下一个执行的协程。

　　测试用例：

import asyncio
import time
from random import randint

@asyncio.coroutine
def StartState():
    print("Start State called 
")
    input_value = randint(0,1)
    time.sleep(1)
    print("I am StartState.input_value is %s" %input_value)
    if (input_value == 0):
        result = yield from State2(input_value)
    else:
        result = yield from State1(input_value)
    print("Resume of the Transition : 
Start State calling %s" %result)

@asyncio.coroutine
def State1(transition_value):
    outputValue = str("State 1 with transition value = %s 
" %transition_value)
    input_value = randint(0,1)
    time.sleep(1)
    print("...Evaluation...")
    print("I am State1.input_value is %s" %input_value)
    if input_value == 0:
        result = yield from State3(input_value)
    else:
        result = yield from State2(input_value)
    result = "State 1 calling %s" %result
    return outputValue + str(result)

@asyncio.coroutine
def State2(transition_value):
    outputValue = str("State 2 with transition value = %s 
" %transition_value)
    input_value = randint(0,1)
    time.sleep(1)
    print("...Evaluation...")
    print("I am State2.input_value is %s" %input_value)
    if input_value == 0:
        result = yield from State1(input_value)
    else:
        result = yield from State3(input_value)
    result = "State 2 calling %s" %result
    return outputValue + str(result)

@asyncio.coroutine
def State3(transition_value):
    outputValue = str("State 3 with transition value = %s 
" %transition_value)
    input_value = randint(0,1)
    time.sleep(1)
    print("...Evaluation...")
    print("I am State3.input_value is %s" %input_value)
    if input_value == 0:
        result = yield from State1(input_value)
    else:
        result = yield from EndState(input_value)
    result = "State 1 calling %s" %result
    return outputValue + str(result)

@asyncio.coroutine
def EndState(transition_value):
    outputValue = str("End State with transition value = %s 
" %transition_value)
    print("I am EndState.outputValue is %s" %outputValue)
    print("...Stop Computation...")

    return outputValue

if __name__ == "__main__":
    print("Finite State Machine simulation with Asyncio Coroutine")
    loop = asyncio.get_event_loop()
    loop.run_until_complete(StartState())

　　上述代码为使用Asyncio的协程来模拟有限状态机(一个数学模型，不仅在工程领域应用广泛，在科学领域也很著名)。模拟的状态机如下：

　　

　　系统有四个状态，0和1是状态机可以从一个状态到另一个状态的值，这个过程叫转换。

　　运行结果(结果不唯一)：

　　

　　每一个状态都由一个装饰器装饰：@asyncio.coroutine

　　通过yield from命令调用下一个协程。

　　启动事件循环：

if __name__ == "__main__":
    print("Finite State Machine simulation with Asyncio Coroutine")
    loop = asyncio.get_event_loop()
    loop.run_until_complete(StartState())

5、使用Asyncio控制任务

　　Asyncio是用来处理事件循环中的异步进程和并发任务执行的。它还提供了asyncio.Task()类，可以在任务中使用协程。它的作用是在同一事件循环中，运行某一个任务的同时可以并发地运行多个任务。当协程被包在任务中，它会自动将任务和事件循环连接起来，当事件循环启动的时候，任务自动运行。这样就提供了一个可以自动驱动协程的机制。

　　Asyncio模块为我们提供了asyncio.Task(coroutine)方法来处理计算任务，它可以调度协程的执行。任务对协程对象在事件循环的执行负责。如果被包裹的协程要从future yield，那么任务会被挂起，等待future的计算结果。

　　当future计算完成，被包裹的协程将会拿到future返回的结果或异常(exception)继续执行。另外，需要注意的是事件循环一次只能运行一个任务，除非还有其它事件循环在不同的线程并行运行，此任务才有可能和其他任务并行。当一个任务在等待future执行的期间，事件循环会运行一个新的任务。

　　测试用例：

import asyncio

@asyncio.coroutine
def factorial(number):
    f = 1
    for i in range(2, number + 1):
        print("Asyncio.Task: Compute factorial(%s)" %i)
        yield from asyncio.sleep(0.5)
        f *= i
    print("Asyncio.Task - factorial(%s) = %s" %(number, f))

@asyncio.coroutine
def fibonacci(number):
    a,b = 0,1
    for i in range(number):
        print("Asyncio.Task: Compute fibonacci(%s)" %i)
        yield from asyncio.sleep(0.5)
        a, b = b, a+b
    print("Asyncio.Task - fibonacci(%s) = %s" %(number, a))

@asyncio.coroutine
def binomialCoeff(n, k):
    result = 1
    for i in range(1, k+1):
        result = result * (n-i+1)/i
        print("Asyncio.Task:Compute binomialCoeff(%s)" %i)
        yield from asyncio.sleep(0.5)
    print("Asyncio.Task - binomialCoeff(%s, %s) = %s" %(n, k, result))

if __name__ == "__main__":
    tasks = [asyncio.Task(factorial(10)), asyncio.Task(fibonacci(10)), asyncio.Task(binomialCoeff(20, 10))]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

　　执行结果：

Asyncio.Task: Compute factorial(2)
Asyncio.Task: Compute fibonacci(0)
Asyncio.Task:Compute binomialCoeff(1)
Asyncio.Task: Compute factorial(3)
Asyncio.Task: Compute fibonacci(1)
Asyncio.Task:Compute binomialCoeff(2)
Asyncio.Task: Compute factorial(4)
Asyncio.Task: Compute fibonacci(2)
Asyncio.Task:Compute binomialCoeff(3)
Asyncio.Task: Compute factorial(5)
Asyncio.Task: Compute fibonacci(3)
Asyncio.Task:Compute binomialCoeff(4)
Asyncio.Task: Compute factorial(6)
Asyncio.Task: Compute fibonacci(4)
Asyncio.Task:Compute binomialCoeff(5)
Asyncio.Task: Compute factorial(7)
Asyncio.Task: Compute fibonacci(5)
Asyncio.Task:Compute binomialCoeff(6)
Asyncio.Task: Compute factorial(8)
Asyncio.Task: Compute fibonacci(6)
Asyncio.Task:Compute binomialCoeff(7)
Asyncio.Task: Compute factorial(9)
Asyncio.Task: Compute fibonacci(7)
Asyncio.Task:Compute binomialCoeff(8)
Asyncio.Task: Compute factorial(10)
Asyncio.Task: Compute fibonacci(8)
Asyncio.Task:Compute binomialCoeff(9)
Asyncio.Task - factorial(10) = 3628800
Asyncio.Task: Compute fibonacci(9)
Asyncio.Task:Compute binomialCoeff(10)
Asyncio.Task - fibonacci(10) = 55
Asyncio.Task - binomialCoeff(20, 10) = 184756.0

　　上述例子定义了三个线程，factorial，fibonacci，binomialCoeff，每一个都带有asyncio.coroutine装饰器：

　　将三个task放入到一个list中：

tasks = [asyncio.Task(factorial(10)), asyncio.Task(fibonacci(10)), asyncio.Task(binomialCoeff(20, 10))]

　　通过run_until_complete并行运行三个协程，asyncio.wait(tasks)表示运行直到所有给定的协程都完成。

　　最后关闭事件循环：loop.close()

6、使用Asyncio和Futures

　　Asyncio模块的另一个重要的组件是Futures。它和concurrent.futures.Futures很像，但是针对Asyncio的事件循环做了很多定制。asyncio.Futures类代表还未完成的结果，有可能是一个Exception，所以综合来说，它是一种抽象的代表还没有做完的事情。

　　实际上，必须处理一些结果的回调函数被加入到了这个类的实例中。

　　基本方法：

　　- cancel()：取消future的执行，调度回调函数

　　- result()：返回future代表的结果

　　- exception()：返回future中的Exception

　　- add_done_callback(fn)：添加一个回调函数，当future执行的时候会调用这个回调函数

　　- remove_done_callback(fn)：从call when done列表中移除所有的callback的实例

　　- set_result(result)：将future标为执行完成，并且设置result的值

　　- set_exception(exception)：将future标为执行完成，并设置Exception

　　测试用例：

# coding : utf-8
import asyncio
import sys

@asyncio.coroutine
def first_coroutine(future, n):
    # 计算前n个数的和
    count = 0
    for i in range(1, n+1):
        count = count + i
    print("first yield")
    yield from asyncio.sleep(2)
    print("first_coroutine finished")
    # 将future标记为已完成，并设置result的值
    future.set_result("first coroutine （sum of n integers） result = %s" %str(count))

@asyncio.coroutine
def second_coroutine(future, n):
    count = 1
    for i in range(2, n+1):
        count *= i
    print("second yield")
    yield from asyncio.sleep(1)
    print("second_coroutine finished")
    future.set_result("second coroutine (factorial) result = %s" %str(count))

def got_result(future):
    # 获取future的set_result结果
    print(future.result())

if __name__ == "__main__":
    N1 = int(sys.argv[1])
    N2 = int(sys.argv[2])
    loop = asyncio.get_event_loop()
    future1 = asyncio.Future()
    future2 = asyncio.Future()
    tasks = [first_coroutine(future1, N1), second_coroutine(future2, N2)]

    # 添加回调函数
    future1.add_done_callback(got_result)
    future2.add_done_callback(got_result)
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

　　运行结果：