在Python中使用asyncio进行异步编程

Posted 2021-02-03 erhuabushuo

对于来自JavaScript编码者来说，异步编程不是什么新东西，但对于Python开发者来说，async函数和future(类似JS的promise)可不是那么容易能理解的。

Concurrency vs Parallelism

Concurrency和Parallelism听起来一样，但在实际编程里它们有着较大的不同。

想象下你在做饭的时候写书，看起来好像你在同一时间做两件事情，实际你只是在两项事情中相互切换，当你在等水开的时候你就可以去写书，但你切菜时你要暂停写作。这就叫做concurrency。唯一一种使用parallel做这两项工作的办法就是得有两个人，一个人写作，一个人做饭，这就是多核CPU的工作方式了。

为什么asyncio

异步编程允许你在单个线程中并发执行代码。对比多线程处理方式，该方式由你来决定如何由一个任务切换到另一个任务，tasks之间共享数据也更加容易和简单。

    def queue_push_back(x):
        if len(list) < max_size:
            list.append(x)

如果我们在多线程执行上面的额代码，有可能第二行代码在同一时间被执行，那么同一时间就有两个元素被加入到列表中，那实际列表长度就会操作max_size。

另一个异步编程的好处是内存使用。每次一个新的线程创建，也需要开辟一些新内存用来进行上下文切换。如果使用了异步编程，这在单线程中就不存在该问题。

如何在Python编程async代码

Asyncio包含三个主要组件：coroutine, event loop和future

Coroutine

coroutine是异步函数，通过在函数定义def前使用async关键字。

async def my_task(args):
    pass

my_coroutine = my_task(args)

我们使用了async关键字定义了一个函数，该函数并没有执行，返回了一个coroutine对象。

有两种从一个coroutine中获取异步函数的结果

第一种使用await关键字，仅只能在async函数中用来等待coroutine结束返回结果

result = await my_task(args)

第二种是将它加入到event loop中，接下来我们做详尽讨论。

Event loop

event loop对象负责执行异步代码以及决定异步函数如何进行切换。在创建了event loop后，我们就可以添加多个coroutines给它，coroutines将会调用了run_until_complete或者run_forever执行。

# create loop
loop = asyncio.new_event_loop()
# add coroutine to the loop
future = loop.create_task(my_coroutine)
# stop the program and execute all coroutine added
# to the loop concurrently
loop.run_until_complete(future)
loop.close()

Future

future类似一个占位对象用来存放异步函数结果，提供函数状态。当coroutine添加到event lop时创建future.有两种方式创建：

future1 = loop.create_task(my_coroutine)
# or
future2 = asyncio.ensure_future(my_coroutine)

第一个方法是增加一个coroutine到loop中，返回一个task，它是future的子类。第二种方法非常类似，它接收一个coroutine，并加入到了默认loop中，唯一的区别是，它也可以接收一个future参数，它将不会做任何事情，直接将futrue返回。

一个简单的程序

import asyncio

async def my_task(args):
    pass

def main():
    loop = asyncio.new_event_loop()
    coroutine1 = my_task()
    coroutine2 = my_task()
    task1 = loop.create_task(coroutine1)
    task2 = loop.create_task(coroutine2)
    loop.run_until_complete(asycnio.wait([task1, task2]))
    print('task1 result:', task1.result())
    print('task2 result:', task2.result())
    loop.close()

就让如你所看见的，我们在执行异步函数前需要先创一个coroutine，然后我们将创建future/task，把它添加到event loop。到现在病没有如何的异步函数被执行，只有当我们调用loop.run_until_completed，event loop开始执行所有的通过loop.createt_task或者asyncio.ensure_future添加的coroutines。loop.run_until_completed将会阻塞应用程序，仅当所有的future执行完毕后。在本例中，我们使用asyncio.wait()创建future，当传递的所有future执行完后我们就获取到了future所有的结果。

异步函数

有一件事需要注意的是在Python中使用async声明的函数并不意味着函数会并发执行。如果使用一个普通函数，在前面加入async关键字，event loop并不会中断你的函数去执行另一个coroutine。允许event loop进行切换coroutine相当简单，使用await关键字就会允许event loop可以切换其他注册到loop中的coroutine。

import asyncio

async def print_numbers_async1(n, prefix):
    for i in range(n):
        print(prefix, i)

async def print_numbers_async2(n, prefix):
    for i in range(n):
        print(prefix, i)
        if i % 5 == 0:
            await asyncio.sleep(0)
            
loop1 = asyncio.new_event_loop()
count1_1 = loop1.create_task(print_numbers_async1(10, 'c1_1'))
count2_1 = loop1.create_task(print_numbers_async1(10, 'c2_1'))
loop1.run_until_complete(asyncio.wait([count1_1, count2_1]))
loop1.close()

loop2 = asyncio.new_event_loop()
count1_2 = loop2.create_task(print_numbers_async2(10, 'c1_2'))
count2_2 = loop2.create_task(print_numbers_async2(10, 'c2_2'))
loop2.run_until_complete(asyncio.wait([count1_2, count2_2]))
loop2.close()

如果我们执行该代码，我们可以看到loop1将会在c1_1完全执行完后才去执行c2_1。而在loop2每打印五个数值后就会进行切换。

真实案例

现在我们Python中最进本的异步编程，现在让我们写一个真实例子，我们从互联网下载一系列页面，并打印出开头三行。

import aiohttp
import asyncio

async def print_preview(url):
    # connect to the server
    async with aiohttp.ClientSession() as session:
        # create get request
        async with session.get(url) as response:
            # wait for response
            response = await response.text()

            # print first 3 not empty lines
            count = 0
            lines = list(filter(lambda x: len(x) > 0, response.split('
')))
            print('-'*80)
            for line in lines[:3]:
                print(line)
            print()

def print_all_pages():
    pages = [
        'http://textfiles.com/adventure/amforever.txt',
        'http://textfiles.com/adventure/ballyhoo.txt',
        'http://textfiles.com/adventure/bardstale.txt',
    ]

    tasks =  []
    loop = asyncio.new_event_loop()
    for page in pages:
        tasks.append(loop.create_task(print_preview(page)))

    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

if __name__ == "__main__":
    print_all_pages()

这里的代码也很容易理解，我们使用异步函数下载URL，并且打印了前三行。然后我们创建了一个函数用来构建一个页面了列表，交给print_preview去执行，将coroutine加入到loop，把future放到了一个列表中 ，我们执行event loop，在所有coroutine执行完后程序结束。

异步生成器

最后我想谈谈的是异步生成器。要实现一个异步生成器相当简单：

import asyncio
import math
import random

async def is_prime(n):
    if n < 2:
        return True

    for i in range(2, n):
        await asyncio.sleep(0)
        if n % i == 0:
            return False

    return True


async def prime_generator(n_prime):
    counter = 0
    n = 0
    while counter < n_prime:
        n += 1
        prime = await is_prime(n)
        if prime:
            yield n
            counter += 1
    
async def check_email(limit):
    for i in range(limit):
        if random.random() > 0.8:
            print('1 new email')
        else:
            print('0 new email')
        await asyncio.sleep(2)

async def print_prime(n):
    async for prime in prime_generator(n):
        print('new prime number found:', prime)


def main():
    loop = asyncio.new_event_loop()
    prime = loop.create_task(print_prime(3000))
    email = loop.create_task(check_email(10))
    loop.run_until_complete(asyncio.wait([prime, email]))
    loop.close()

if __name__ == "__main__":
    main()

异常处理

在coroutine内部抛出异常时并不会中断应用程序，如果你没有处理异常的话你将看到类似如下错误：

Task exception was never retrieved

有两个方法来修正，在获取future结果时捕获异常，或者在future调用exception方法.

深入了解

现在你已经了解如何使用asyncio编写并发代码，如果你想深入了解的话，查看官方文档。