《asyncio 并发编程》（第四章）——并发网络请求

第三章中，我们构建了一个基本的回显服务器，在本章中，我们将学习一个名为aiohttp的异步库，以实现同时发出数百个请求，并同时运行这些请求。

1 aiohttp

aiohttp 是一个为 Python 提供异步 HTTP 客户端和服务器功能的库。它基于 asyncio 异步编程框架，允许开发者编写高并发性能的应用程序。

使用 aiohttp 可以进行非阻塞式的网络操作，这对于需要同时处理大量网络连接的应用来说尤为重要。

在使用之前，首先要学习关于异步上下文管理器的新语法，可以允许我们获取和关闭HTTP会话。

1.1 异步上下文管理器

在之前的python编程中，我们会使用try...finally...进行抛出异常然后进行操作，为了更简洁，Python有一种上下文管理器来将二者一起抽象：

with open('example.txt') as file:
	lines = file.readlines()
	
'''
以只读模式打开当前目录下的 example.txt 文件,
在 with 语句块内读取该文件的所有行到一个列表 lines 中,
且无论是否发生异常，都会自动关闭文件。
'''

这种适用于同步语法，在异步情况下，就不起作用了，所以引入了一种异步上下文管理器，语法基本相同，只是会采用async with

异步上下文管理器是实现两个特殊协程方法的类，_aenter_是异步获取资源，_aexit_是关闭该资源

class MyAsyncContextManager:
    async def __aenter__(self):
        # 在进入运行时上下文之前调用
        # 可能会执行一些异步准备工作
        print("Entering context.")
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 在离开运行时上下文之后调用
        # 这里可以处理异常或清理工作
        print("Exiting context.")

# 使用上述异步上下文管理器
async def main():
    async with MyAsyncContextManager() as manager:
        # 使用上下文管理器提供的资源
        print("Inside the context.")

接下来我们将写一个等待客户端连接的异步上下文管理器

1.2 等待客户端连接的异步上下文管理器

import asyncio
import socket
from types import TracebackType
from typing import Optional, Type

class ConnectedSocket:
    def __init__(self, server_socket):
        self._connection = None
        self._server_socket = server_socket
        

    async def __aenter__(self):
        print("Entering context manager, waiting for connection...")
        loop = asyncio.get_event_loop()
        connection, address = await loop.sock_accept(self._server_socket)
        self._connection = connection
        print("Accepted connection from", connection.getpeername())
        return self._connection
    
    async def __aexit__(self, 
                      exc_type: Optional[Type[BaseException]], 
                      exc_val: Optional[BaseException], 
                      exc_tb: Optional[TracebackType]):
        print("Exiting context manager, closing connection...")
        self._connection.close()
        print("Connection closed")
        
async def main():
    loop = asyncio.get_event_loop()

    server_socket = socket.socket()
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_address = ('localhost', 8080)
    server_socket.setblocking(False)
    server_socket.bind(server_address)
    server_socket.listen()

    async with ConnectedSocket(server_socket) as connection:
        connection.setblocking(False)
        data = await loop.sock_recv(connection, 1024)
        print(f"Received: {data.decode()}")

if __name__ == '__main__':
    asyncio.run(main())

我们创建了一个ConnectedSocket异步上下文管理器，接受一个服务器套接字，并在__aenter__协程中等待客户端连接，然后在async with 块中，等待客户端发送数据，一旦接收到客户端的数据，__aexit__协程就会运行并关闭连接。最终服务器输出如下：

aiohttp广泛使用异步上下文管理器来获取HTTP会话和连接，但通常情况下，我们不需要自己编写异步上下文管理器。

1.3 使用aiohttp发出web请求

本文使用aiohttp==3.8.1，首先介绍一个概念叫做连接池。

连接池是指在客户端会话（ClientSession）中管理的用于执行 HTTP 请求的一组持久化连接。通过使用连接池，aiohttp 可以重用到特定主机和端口的连接，从而减少建立新连接所需的时间，提高请求效率，并降低资源消耗。如果有必要，也可以手动关闭此功能。

接下来我们将写一个简单的发出aiohttp请求程序

1.3 .1 发出aiohttp请求程序

import aiohttp
import asyncio
from aiohttp import ClientSession
from util import async_timed

# 异步函数：获取指定URL的HTTP状态码
# 参数：
#   - session: aiohttp客户端会话
#   - url: 目标网址
# 返回：HTTP状态码
@async_timed()
async def fetch_status(session: ClientSession, url: str) -> int:
    async with session.get(url) as result:
        return result.status

# 主异步函数：创建客户端会话并发起HTTP请求
@async_timed()
async def main():
    # 创建aiohttp客户端会话，使用异步上下文管理器确保会话正确关闭
    async with aiohttp.ClientSession() as session:
        url = "https://www.example.com"
        status = await fetch_status(session, url)
        print(f"Status for {url} was {status}")

# 运行异步主函数
asyncio.run(main())

输出如下：

注意，默认情况下，ClientSession最多创建100个连接，若要更改此限制，可以创建一个aiohttp TCPConnector实例，指定最大连接数，并将其传递给ClientSession

1.3.2 使用aiohttp设置超时

可以用aiohttp特定的ClientTimeout来指定超时。

ten_millis = aiohttp.ClientTimeout(total=.01)
# 超过了 10 毫秒，就会触发超时异常

2 并发运行任务及重新访问

在前面我们学习如何创建多个任务来并行运行程序，然而在需要同时发出数百、数千或者更多Web请求的情况下，必然需要一种“循环方法”来简化这个过程

2.1 通过gather执行并发请求

是asyncio 库提供的一个函数，用于并行地调度多个协程，并且等待它们全部完成。

它接收多个 awaitable 对象（如协程）作为参数，并返回一个包含所有传入的 awaitable 对象的结果的列表。

下面是一个例子，使用gather同时发出100个请求，并获取每个响应的状态码：

import aiohttp
import asyncio
from aiohttp import ClientSession
from util import async_timed

# 异步函数：获取URL的HTTP状态码
# 使用async_timed装饰器记录执行时间
@async_timed()
async def fetch_status(session: ClientSession, url: str) -> int:
    # 使用异步上下文管理器发送GET请求
    async with session.get(url) as result:
        return result.status

# 主函数：并发发送100个HTTP请求
@async_timed()
async def main():
    # 创建一个异步HTTP客户端会话
    async with aiohttp.ClientSession() as session:
        # 生成100个相同的URL
        urls = ["https://www.example.com" for _ in range(100)]
        # 为每个URL创建一个请求任务
        requests = [fetch_status(session, url) for url in urls]
        # 使用asyncio.gather并发执行所有请求
        status_codes = await asyncio.gather(*requests)
        # 打印所有请求的状态码
        print(status_codes)

# 启动异步事件循环并运行主函数
asyncio.run(main())

输出结果


如果你在上述代码中不使用 asyncio.gather()，而是选择使用普通的循环来依次等待每个协程完成，那么所有的请求将会被顺序执行而不是并发执行。这意味着每一个 fetch_status 调用必须等待前一个调用完成之后才能开始，这会导致总的执行时间显著增加。

注意：每个awaitable对象的结果可能不是按照顺序调用完成的，但是gather会保证他们按照顺序返回他们的结果

2.1.1 处理gather中的异常、

当发出网络请求时，不一定能得到一个返回值，有可能会得到一个异常asyncio.gather提供了一个可选参数return_exceptions，可以指定如何处理异常：

• 默认行为（return_exceptions=False）： 如果任何一个 awaitable对象抛出了异常，asyncio.gather() 会立即停止处理，并且不会等待其他 awaitable完成。它会直接传播第一个遇到的异常，导致其他正在执行的 awaitables 可能会被取消，除非处理了这个异常。
• 设置 return_exceptions=True： 如果有任何 awaitable抛出异常，异常不会被抛出而是作为结果返回。这样，即使某些任务失败了，其他成功完成的任务的结果仍然可以获取到，程序也可以继续执行。

下面我将举一个例子：
1.采用默认的False情况

import aiohttp
import asyncio
from aiohttp import ClientSession
from util import async_timed

# 异步函数：获取URL的HTTP状态码
# 使用async_timed装饰器记录执行时间
@async_timed()
async def fetch_status(session: ClientSession, url: str) -> int:
    # 使用异步上下文管理器发送GET请求
    async with session.get(url) as result:
        return result.status


@async_timed()
async def main():
    async with aiohttp.ClientSession() as session:
        urls = ["https://www.example.com", "python://example.com"]
        requests = [fetch_status(session, url) for url in urls]
        status_codes = await asyncio.gather(*requests)#默认情况
        print(status_codes)

asyncio.run(main())

输出结果：

会报错，但是也能看到另一个网站的正常运行。
2.采用True情况

import aiohttp
import asyncio
from aiohttp import ClientSession
from util import async_timed

# 异步函数：获取URL的HTTP状态码
# 使用async_timed装饰器记录执行时间
@async_timed()
async def fetch_status(session: ClientSession, url: str) -> int:
    # 使用异步上下文管理器发送GET请求
    async with session.get(url) as result:
        return result.status


@async_timed()
async def main():
    async with aiohttp.ClientSession() as session:
        urls = ["https://www.example.com", "python://example.com"]
        requests = [fetch_status(session, url) for url in urls]
        status_codes = await asyncio.gather(*requests,return_exceptions=True)#启动return_exceptions=True
        print(status_codes)

asyncio.run(main())

输出结果：

一切正常运行，没有抛出异常，而是将异常返回，正常运行了所有程序。

gather函数也有一些潜在的缺点：

1. 缺乏细粒度控制： asyncio.gather() 以一种“全有或全无”的方式运行任务集合。这意味着它不提供对单个任务的中途取消或状态监控的支持。
2. 调试困难： 由于所有任务都是并发执行的，如果其中某些任务出现了问题，追踪具体的错误源头可能会更加困难。
3. 不够灵活： 必须等待所有的协程执行完成，然后才能处理结果，例如一个请求需要100ms，另一个需要20s，那么在处理100ms的请求之前，必须等待20s的请求结束后才能处理

2.2 在请求完成时立即处理

为了解决不够灵活这个问题，asyncio公开了一个as_completed的API函数，该方法接收一个等待列表并返回一个future迭代器，然后迭代这些future，并等待它们中的每一个完成，当结果可用时，就立刻去处理它们。

为了说明是如何工作的，这里我模拟一个请求快速完成，另一个请求需要更多时间的情况。将在fetch_status函数中添加一个delay参数，然后调用sleep来模拟一个长时间的请求

import asyncio

async def task(name, duration):
    print(f"Task {name} starting and will take {duration}s to complete.")
    await asyncio.sleep(duration)  # 使用sleep来模拟耗时操作
    print(f"Task {name} finished.")
    return f"{name} - Done"

async def main():
    # 创建一个异步HTTP客户端会话和一些模拟任务
    tasks = [
        task('Short', 1),  # 短任务，模拟2秒的操作
        task('Long', 5)    # 长任务，模拟5秒的操作
    ]
    
    # 使用asyncio.as_completed来迭代已经完成的任务
    for future in asyncio.as_completed(tasks):
        result = await future
        print(result)

# 启动异步事件循环并运行主函数
asyncio.run(main())

输出结果：

会发现当Short完成后，立刻出现Short - Done，过了4秒后，才出现Long完成，然后出现Long - Done。

我们可以把该函数结合到异常处理中，一旦处理异常，则可以立刻对异常进行处理。

2.2.1 as_completed 超时

任何Web请求都有可能超时，as_completed提供了一个timeout参数来处理超时情况，如果在指定的时间内某个 协程没有完成，就会抛出 asyncio.TimeoutError 异常。

import asyncio
import aiohttp
from aiohttp import ClientSession
from util import async_timed

# 异步函数：获取URL的HTTP状态码
async def fetch_status(session: aiohttp.ClientSession, url: str, delay: int = 0) -> int:
    # 模拟延迟
    await asyncio.sleep(delay)
    # 使用异步上下文管理器发送GET请求
    async with session.get(url) as result:
        return result.status

@async_timed()
async def main():
    async with aiohttp.ClientSession() as session:
        fetchers = [
            fetch_status(session, 'https://example.com', 1),# 模拟反应1秒
            fetch_status(session, 'https://example.com', 10),# 模拟反应10秒
            fetch_status(session, 'https://example.com', 10)# 模拟反应10秒
        ]

        for done_task in asyncio.as_completed(fetchers, timeout=2):
            try:
                result = await done_task
                print(result)
            except asyncio.TimeoutError:
                print('We got a timeout error!')

        for task in asyncio.tasks.all_tasks():
            print(task)

asyncio.run(main())

输出结果：

观察结果可以发现返回一个状态码和两个超时错误，还发现两个fetch还在运行Task pending name='Task-4' coro=<fetch_status() running......

所以有两个缺点：

1. 当得到结果时，没有办法快速了解我们正在等待哪个协程或任务，因为运行顺序是不确定的。
2. 虽然抛出了异常，但是任务还是在后台运行，如果想取消，则和第一个问题一样，我们无法确定要取消哪一个任务

2.3 使用wait进行细粒度控制

之前在讨论gather和as_completed时就说过，他俩的细粒度控制不足，无法对单个任务或者协程进行监控，为了解决这个问题，我们可以使用asyncio的wait来解决。

asyncio.wait() 是 Python asyncio 库中用于并发调度协程或等待多个 Future 对象完成的一个函数。它允许你同时等待多个协程或 Future，并且提供了灵活性来决定是等待所有任务完成还是只要有任何一个任务完成就继续执行。

asyncio.wait() 有三个参数：

1. futures: 这是一个包含要等待的 Future 或协程对象的集合。
2. timeout: 可选参数，设置等待的最大秒数。如果超时发生，未完成的任务将被留在 pending 集合中。
3. return_when: 决定函数何时返回，包含三个选项：

• 默认值为 ALL_COMPLETED，即等待所有给定的 Future 完成。
• FIRST_COMPLETED: 函数将在任意 Future 完成或被取消时立即返回。
• FIRST_EXCEPTION: 函数将在任意 Future 抛出异常时立即返回。如果没有 Future 抛出异常，则此选项的行为与 ALL_COMPLETED 相同。

asyncio.wait() 有两个返回集合：

• done: 包含已经完成（包括正常完成、因异常完成或被取消）的所有 Future 和协程。
• pending: 包含尚未完成的所有 Future 和协程。

2.3.1 等待所有任务完成

默认值是ALL_COMPLETED，即等待所有给定的 Future 完成。

import asyncio
import aiohttp
from aiohttp import ClientSession
from util import async_timed

# 异步函数：获取URL的HTTP状态码
async def fetch_status(session: aiohttp.ClientSession, url: str, delay: int = 0) -> int:
    # 模拟延迟
    await asyncio.sleep(delay)
    # 使用异步上下文管理器发送GET请求
    async with session.get(url) as result:
        return result.status

@async_timed()
async def main():
    async with aiohttp.ClientSession() as session:
        fetchers = [
            asyncio.create_task(fetch_status(session, 'https://example.com')),
            asyncio.create_task(fetch_status(session, 'https://example.com'))
        ]
        done, pending = await asyncio.wait(fetchers)

    print(f'Done task count: {len(done)}')
    print(f'Pending task count: {len(pending)}')

    for done_task in done:
        result = await done_task
        print(result)

asyncio.run(main())

输出结果：

因为两个url都是可以访问的，所以2个都是完成的。

如果有异常抛出，它不会入gather一样在调用中被抛出，而是获得完成集合和未完成集合，这样我们便可以结合wait进行异常处理，下面展示如何使用wait进行异常处理。

import asyncio
import logging
import aiohttp
from util import async_timed
# 异步函数：获取URL的HTTP状态码
async def fetch_status(session: aiohttp.ClientSession, url: str, delay: int = 0) -> int:
    # 模拟延迟
    await asyncio.sleep(delay)
    # 使用异步上下文管理器发送GET请求
    async with session.get(url) as result:
        return result.status

@async_timed()
async def main():
    async with aiohttp.ClientSession() as session:
        good_request = fetch_status(session, 'https://www.example.com')
        bad_request = fetch_status(session, 'python://bad')

        fetchers = [
            asyncio.create_task(good_request),
            asyncio.create_task(bad_request)
        ]

        done, pending = await asyncio.wait(fetchers)

        print(f'Done task count: {len(done)}')
        print(f'Pending task count: {len(pending)}')

        for done_task in done:
            # result = await done_task will throw an exception
            if done_task.exception() is None:
                print(done_task.result())
            else:
                logging.error("Request got an exception", exc_info=done_task.exception())

asyncio.run(main())

输出结果：

遍历已完成的任务，检查每个任务是否抛出异常。如果没有异常，则打印任务的结果；如果有异常，则记录错误信息。

2.3.2 观察异常

ALL_COMPLETED如同gather一样，只有所有协程运行完了，才能看到异常信息，这必然不是我们想要得，这里将介绍FIRST_EXCEPTION，使用这个选项时，会得到两种不同的行为，取决于任务是否抛出异常：

1. 任何可等待对象都没有异常： 则该选项等效于ALL_COMPLETED
2. 任务中存在一个或者多个异常： 如果其中任何一个任务抛出异常，wait函数将立即返回。此时，完成集合会包含所有已经成功完成的协程，以及任何有异常的协程。在这种情况下，完成集合至少会有一个失败的任务，也可能包含一些已经成功完成的任务。挂起的集合可能是空的，但也可能包含仍在运行的任务。然后，可以根据需要使用这个待处理的集合来管理当前仍在运行的任务。
   下面将举一个例子，当有两个长时间运行的Web请求时，当一个协程发生异常，则立刻取消正在运行的请求：

import aiohttp
import asyncio
import logging
from util import async_timed

# 异步函数：获取URL的HTTP状态码
async def fetch_status(session: aiohttp.ClientSession, url: str, delay: int = 0) -> int:
    # 模拟延迟
    await asyncio.sleep(delay)
    # 使用异步上下文管理器发送GET请求
    async with session.get(url) as result:
        return result.status

@async_timed()
async def main():
    async with aiohttp.ClientSession() as session:
        fetchers = [
            asyncio.create_task(fetch_status(session, 'python://bad.com')),# 这是错误的URL
            asyncio.create_task(fetch_status(session, 'https://www.example.com', delay=3)),
            asyncio.create_task(fetch_status(session, 'https://www.example.com', delay=3))
        ]

        done, pending = await asyncio.wait(fetchers, return_when=asyncio.FIRST_EXCEPTION)

        print(f'Done task count: {len(done)}')
        print(f'Pending task count: {len(pending)}')
        for done_task in done:
            if done_task.exception() is None:
                print(done_task.result())
            else:
                logging.error("Request got an exception", exc_info=done_task.exception())

        for pending_task in pending:
            pending_task.cancel()# 取消未完成的任务

asyncio.run(main())

输出结果：

Done task count: 1
Pending task count: 2
ERROR:root:Request got an exception
Traceback (most recent call last):
  File "/home/caser/code/asyncioStudy/aiohttp_sendout.py", line 11, in fetch_status
    async with session.get(url) as result:
  File "/home/caser/miniconda3/envs/pythonProject/lib/python3.8/site-packages/aiohttp/client.py", line 1138, in __aenter__
    self._resp = await self._coro
  File "/home/caser/miniconda3/envs/pythonProject/lib/python3.8/site-packages/aiohttp/client.py", line 535, in _request
    conn = await self._connector.connect(
  File "/home/caser/miniconda3/envs/pythonProject/lib/python3.8/site-packages/aiohttp/connector.py", line 542, in connect
    proto = await self._create_connection(req, traces, timeout)
  File "/home/caser/miniconda3/envs/pythonProject/lib/python3.8/site-packages/aiohttp/connector.py", line 907, in _create_connection
    _, proto = await self._create_direct_connection(req, traces, timeout)
  File "/home/caser/miniconda3/envs/pythonProject/lib/python3.8/site-packages/aiohttp/connector.py", line 1146, in _create_direct_connection
    assert port is not None
AssertionError
finished main in 0.0007 second(s)

1. Done task count: 1：表示有一个任务已经完成。在这个上下文中，这意味着在设置return_when=asyncio.FIRST_EXCEPTION的情况下，一旦遇到第一个异常，asyncio.wait()就会返回，标记那些已完成的任务。在这里，只有一个任务完成了（实际上是由于抛出了异常而结束的）。
2. Pending task count: 2：表示有两个任务尚未完成。这两个任务是被安排执行但因第一个异常的发生而未被执行到或者正在等待中的任务。由于设置了在出现第一个异常时不再等待其他任务完成，所以这两个任务被标记为pending。
3. finished main in 0.0007 second(s)：这说明整个main函数的执行时间非常短（大约0.0007秒），是因为程序几乎立刻就遇到了第一个异常（即访问python://bad.com失败），然后快速地处理了异常情况并退出了。

说明当出现异常时，非常迅速地就做出了响应

2.3.3 当任务完成时处理结果

FIRST_EXCEPTION和ALL_COMPLETED都有一个缺点，在协程成功且不抛出异常的情况下，必须等待所有协程执行完成，如果想要在协程成功完成后立即处理，就需要使用到另一个参数FIRST_COMPLETED。

此选项将在wait协程至少有一个结果时立即返回，即可以是失败的协程，也可以是成功的协程。

下面的代码是协程完成后立即处理，主要功能是并发地发起多个HTTP请求，并在第一个请求完成时打印已完成任务的数量、未完成任务的数量以及已完成任务的结果。

import asyncio
import aiohttp
from util import async_timed

# 异步函数：获取URL的HTTP状态码
async def fetch_status(session: aiohttp.ClientSession, url: str, delay: int = 0) -> int:
    # 模拟延迟
    await asyncio.sleep(delay)
    # 使用异步上下文管理器发送GET请求
    async with session.get(url) as result:
        return result.status

@async_timed()
async def main():
    async with aiohttp.ClientSession() as session:
        url = 'https://www.example.com'
        fetchers = [
            asyncio.create_task(fetch_status(session, url)),
            asyncio.create_task(fetch_status(session, url)),
            asyncio.create_task(fetch_status(session, url))
        ]

        done, pending = await asyncio.wait(fetchers, return_when=asyncio.FIRST_COMPLETED)

        print(f'Done task count: {len(done)}')
        print(f'Pending task count: {len(pending)}')

        for done_task in done:
            print(await done_task)

asyncio.run(main())

输出结果：

如果还要继续处理其余的结果，可以采用上面的模式去循环Pending任务，直到它们为空，下面是示例代码：

import asyncio
import aiohttp
from util import async_timed

# 异步函数：获取URL的HTTP状态码
async def fetch_status(session: aiohttp.ClientSession, url: str, delay: int = 0) -> int:
    # 模拟延迟
    await asyncio.sleep(delay)
    # 使用异步上下文管理器发送GET请求
    async with session.get(url) as result:
        return result.status

@async_timed()
async def main():
    async with aiohttp.ClientSession() as session:
        url = 'https://www.example.com'
        pending = [
            asyncio.create_task(fetch_status(session, url)),
            asyncio.create_task(fetch_status(session, url)),
            asyncio.create_task(fetch_status(session, url))
        ]

        while pending:
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)

            print(f'Done task count: {len(done)}')
            print(f'Pending task count: {len(pending)}')

            for done_task in done:
                print(await done_task)

asyncio.run(main())

输出结果：

有的时候会很快，以至于所有请求在很短的时间内就能完成，如下图所示：

只花了0.2秒