python异步编程回顾

淩亂°似流年 2021-11-23 09:38 229阅读 0赞

# 使用concurrent.futures模块 #

该模块包含以下几个类

concurrent.futures.Executor # 这是一个抽象类，提供异步执行的调用方法。
    submit(function,argument)   # 安排某个函数执行。这里的函数是可调用对象，并给定参数。
    map(function,argument)      # 以异步模式使用给定参数来执行函数
    shutdown(Wait=True)         # 向执行器Executor传递释放资源的信号
    concurrent.futures.Future   # 封装一个可调用函数的异步执行。通过向执行器提交任务来实例化Future对象

# 进程池和线程池 #

**concurrent.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 count(number):
        i = None
        for i in range(0,10000000):
            i += 1
        return i*number
    
    def evaluate_item(x):
        result_item = count(x)  # 随便写，只是为了增加一些操作
        print("item " + str(x) + " result: " + str(result_item))
    
    if __name__ == '__main__':
        # *------------------------线性执行-----------------------------
        start_time = time.perf_counter()
        # perf_counter 和 process_time，time.clock(）python3.8将废除
        # https://blog.csdn.net/qq_27283619/article/details/89280974
        for item in number_list:
            evaluate_item(item)
        print("线性执行 task: " + str(time.perf_counter()-start_time) +" s")
    
        # *------------------------线程池执行-----------------------------
        start_time_thread = time.perf_counter()
        with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
            for item in number_list:
                executor.submit(evaluate_item,item)
            print("线程池 先打印，和with有关")
        print("线程池 task: " + str(time.perf_counter()-start_time_thread) +" s")
    
            # *------------------------进程池执行-----------------------------
        start_time_thread = time.perf_counter()
        with concurrent.futures.ProcessPoolExecutor(max_workers=5) as executor:
            for item in number_list:
                executor.submit(evaluate_item,item)
            print("进程池 先打印，和with有关")
        print("进程池 task: " + str(time.perf_counter()-start_time_thread) +" s")
    
    
    """ 计算密集型： 进程 IO密集型 ： 线程，协程(异步） """

# 使用Asyncio实现 时间循环管理 #

python模块的Asyncio提供了管理时间，协程，任务和线程的功能。

*  时间循环(event loop):Asyncio模块 支持**每个进程** 拥有一个 事件循环。
 *  协程(coroutines)：协程可以暂停，等待外部IO处理完成，再继续执行。
 *  Futures:定义了Future对象，类似代表了尚未完成计算的concurrent.futures模块.
 *  任务（tasks）：这是Asyncio的子类，用于封装并管理并行模式下的协程。

**准备工作**，Asyncio提供了用于管理 时间循环的方法

loop = get_event_loop() # 获取当前上下文的时间循环。
    loop.call_later(time_delay,callback,argument) # 给定的时间后，调用某个回调对象
    loop.call_soon(callback,argument) # 立刻调用某个回调对象。call_soon()返回 控制回到时间循环后 会立即调用
    loop.time() #以浮点值的形式返回根据事件循环的内部时钟确定的当前时间
    asyncio.set_event_loop() 将当前上下文的事件循环设置为给定循环。
    asyncio.new_event_loop() 根据此函数的规则创建并返回一个新的事件循环对象
    loop.run_forever() 一直执行，直到调用stop()为止。

import asyncio
    import datetime
    import time
    
    def func_1(end_time,loop):
        print("func_1 called")
        if (loop.time() + 1.0) < end_time:
            loop.call_later(1,func_2,end_time,loop)
        else:
            loop.stop()
    
    def func_2(end_time,loop):
        print("func_2 called")
        if (loop.time() + 1.0) < end_time:
            loop.call_later(1,func_3,end_time,loop)
        else:
            loop.stop()
    
    def func_3(end_time,loop):
        print("func_3 called")
        if (loop.time() + 1.0) < end_time:
            loop.call_later(1,func_1,end_time,loop)
        else:
            loop.stop()
    
    
    def func_4(end_time,loop):
        print("func_4 called")
        if (loop.time() + 1.0) < end_time:
            loop.call_later(1,func_4,end_time,loop)
        else:
            loop.stop()
    
    loop = asyncio.get_event_loop()
    end_loop = loop.time() + 9.0   # 9->8 看效果
    loop.call_soon(func_1,end_loop,loop)
    loop.call_soon(func_4,end_loop,loop)
    loop.run_forever()
    loop.close()

# 使用Asyncio 处理协程 #

有限状态机代码
    https://www.cnblogs.com/aguncn/p/6124928.html

""" 简化栗子 """
    import asyncio
    
    @asyncio.coroutine
    def start():
        print("start")
        yield from end("hello world")
        print("我 start 又回来了")
    
    
    @asyncio.coroutine
    def end(msg):
        print("end")
        print(msg)
        print("end over")
    
    if __name__ == '__main__':
        loop = asyncio.get_event_loop()
        loop.run_until_complete(start())

# 使用asyncio管理任务 #

""" 通过asyncio.Task 并发三个数学函数 """
    
    import asyncio
    
    @asyncio.coroutine
    def factorial(number): # 阶乘
        f = 1
        for i in range(2,number+1):
            print(f"Asyncio.Task:Compute factorial({i})")
            yield from asyncio.sleep(1)
            f *= i
        print(f"Asyncio.Task - factorial({number})={f}")
    
    @asyncio.coroutine
    def fibonacci(number):# 斐波纳契
        a,b = 0,1
        for i in range(number):
            print(f"Asyncio.Task:Compute fibonacci({i})")
            yield from asyncio.sleep(1)
            a,b = b,a+b
        print(f"Asyncio.Task - fibonacci({number}) = {a}")
    
    
    @asyncio.coroutine
    def binomialCoeff(n,k): # 二项式
        result = 1
        for i in range(1,k+1):
            result = result * (n-i+1) / i
            print(f"Asyncio.Task:Compute binomialCoeff({i})")
            yield from asyncio.sleep(1)
        print(f"Asyncio.Task - binomialCoeff({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 和 Futures #

Asyncio模块的另外一个关键组件

它与concurrent.futures.Futures非常相似，但是已经按照asyncio的事件循环的主机制进行了调整。

asyncio.Future 类代表了一个还不可用的结果。因此，它是对尚需完成的任务的抽象表示。

事实上，哪些必须处理结果的回调对象也被算作该类的实例。

**准备工作**

import asyncio
    future = asyncio.Futures()

*  cancel() 取消future 并安排回调对象
 *  result() 返回future所代表的结果
 *  exception() 返回future上设置的异常
 *  add\_done\_callback(fn) 添加一个在future执行时运行的回调对象
 *  remove\_done\_callback(fn) 从 “结束后调用(call when done)” 列表中移除一个回调对象的所有实例
 *  set\_result(result) 将future标记为已完成，并设置其结果
 *  set\_exception(exception) 将future标记为已完成，并设置一个异常

""" Asyncio.Futures """
    
    import asyncio
    import sys
    
    # 求n个整数的和
    @asyncio.coroutine
    def first_coroutine(future,N):
        count = 0
        for i in range(1,N+1):
            count = count + i
        yield from asyncio.sleep(3)
        future.set_result(f"求和 result is {count}")
    
    # 求n的阶乘
    @asyncio.coroutine
    def second_coroutine(future,N):
        count = 1
        for i in range(2,N+1):
            count *= i
        yield from asyncio.sleep(4)
        future.set_result(f"阶乘 result is {count}")
    
    def get_result(future):
        print(future.result())
    
    if __name__ == '__main__':
        N1 = N2 = 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(get_result)
        future2.add_done_callback(get_result)
    
        loop.run_until_complete(asyncio.wait(tasks))
        loop.close()
    
    """ 改变N1 N2值 对调休眠时间 谁先执行完毕，谁先获取。 """

# async def和@asyncio.coroutine #

`async`和`await`是针对coroutine的新语法，要使用新的语法，只需要做两步简单的替换：

1.  把`@asyncio.coroutine`替换为`async`；
2.  把`yield from`替换为`await`。