面试题：网络编程异步操作中兼顾超时与重试的系统架构设计

Question

设计一个完整的后端网络编程系统架构，该架构需在异步编程的基础上，有效融合超时与重试机制，以应对高并发、不稳定网络连接等复杂情况。请详细说明架构中各模块的功能、交互逻辑以及如何通过代码实现这些机制。

Answer 1

后端网络编程系统架构设计

一、架构模块及功能

1. 网络请求模块
   • 功能：负责发起网络请求，接收请求参数（如 URL、请求方法、请求头、请求体等），将请求发送至目标服务器，并接收服务器返回的响应数据。
   • 实现思路：使用异步网络库（如 Python 的 aiohttp 或 Node.js 的 axios 配合 async/await）来实现异步请求。在 Python 中示例代码如下：

import aiohttp

async def send_request(url, method='GET', headers=None, data=None):
    async with aiohttp.ClientSession() as session:
        async with session.request(method, url, headers=headers, json=data) as response:
            return await response.json()

2. 超时控制模块
   • 功能：设置网络请求的超时时间，若请求在设定时间内未完成，终止请求并返回超时错误信息。
   • 实现思路：在异步请求库中，通常支持设置超时参数。以 aiohttp 为例，可在请求时传入 timeout 参数：

import aiohttp
from aiohttp import ClientTimeout

async def send_request_with_timeout(url, method='GET', headers=None, data=None):
    timeout = ClientTimeout(total=5)  # 设置总超时时间为 5 秒
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.request(method, url, headers=headers, json=data) as response:
            return await response.json()

3. 重试机制模块
   • 功能：当网络请求失败（如超时、连接错误、服务器返回错误状态码等）时，根据设定的重试策略进行重试，直至达到最大重试次数或请求成功。
   • 实现思路：通过循环和条件判断实现重试逻辑。在 Python 中示例代码如下：

import aiohttp
from aiohttp import ClientTimeout
import asyncio

async def send_request_with_retry(url, method='GET', headers=None, data=None, max_retries=3, retry_delay=1):
    for attempt in range(max_retries):
        try:
            timeout = ClientTimeout(total=5)
            async with aiohttp.ClientSession(timeout=timeout) as session:
                async with session.request(method, url, headers=headers, json=data) as response:
                    return await response.json()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(retry_delay)
            else:
                raise e

4. 任务调度模块
   • 功能：接收多个网络请求任务，将其按照一定的策略（如优先级）分配到线程池或进程池中执行，提高系统的并发处理能力。
   • 实现思路：在 Python 中可使用 asyncio 库的事件循环结合线程池或进程池来实现。示例代码如下：

import asyncio
import concurrent.futures

executor = concurrent.futures.ThreadPoolExecutor(max_workers=10)

async def schedule_requests(requests):
    loop = asyncio.get_running_loop()
    tasks = [loop.run_in_executor(executor, send_request_with_retry, *args) for args in requests]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

5. 数据处理模块
   • 功能：对网络请求返回的响应数据进行处理，如解析 JSON、XML 数据，提取有用信息，进行数据校验等。
   • 实现思路：根据响应数据的格式，使用相应的解析库。如解析 JSON 数据可使用 Python 的 json 模块，解析 XML 数据可使用 xml.etree.ElementTree 模块。示例代码如下：

import json

def process_response(response_data):
    try:
        data = json.loads(response_data)
        # 进一步处理数据，如提取特定字段
        return data.get('key')
    except json.JSONDecodeError:
        return None

二、交互逻辑

1. 用户发起请求：业务逻辑层调用网络请求模块，传入请求参数。
2. 任务调度：任务调度模块接收请求任务，将其分配到线程池或进程池中执行。
3. 网络请求与超时控制：网络请求模块在执行请求时，超时控制模块开始计时，若请求在超时时间内完成，返回响应数据；否则，抛出超时异常。
4. 重试机制：若网络请求失败，重试机制模块根据设定的重试策略进行重试，直到请求成功或达到最大重试次数。
5. 数据处理：网络请求成功后，数据处理模块对响应数据进行处理，并将处理结果返回给业务逻辑层。

三、代码实现综合示例

以下是一个简单的综合示例，展示上述模块如何协同工作：

import aiohttp
from aiohttp import ClientTimeout
import asyncio
import concurrent.futures
import json


async def send_request(url, method='GET', headers=None, data=None):
    async with aiohttp.ClientSession() as session:
        async with session.request(method, url, headers=headers, json=data) as response:
            return await response.json()


async def send_request_with_timeout(url, method='GET', headers=None, data=None):
    timeout = ClientTimeout(total=5)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.request(method, url, headers=headers, json=data) as response:
            return await response.json()


async def send_request_with_retry(url, method='GET', headers=None, data=None, max_retries=3, retry_delay=1):
    for attempt in range(max_retries):
        try:
            timeout = ClientTimeout(total=5)
            async with aiohttp.ClientSession(timeout=timeout) as session:
                async with session.request(method, url, headers=headers, json=data) as response:
                    return await response.json()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(retry_delay)
            else:
                raise e


executor = concurrent.futures.ThreadPoolExecutor(max_workers=10)


async def schedule_requests(requests):
    loop = asyncio.get_running_loop()
    tasks = [loop.run_in_executor(executor, send_request_with_retry, *args) for args in requests]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results


def process_response(response_data):
    try:
        data = json.loads(response_data)
        return data.get('key')
    except json.JSONDecodeError:
        return None


# 示例使用
async def main():
    requests = [
        ('http://example.com/api1', 'GET'),
        ('http://example.com/api2', 'POST', None, {'param': 'value'})
    ]
    results = await schedule_requests(requests)
    for result in results:
        if isinstance(result, Exception):
            print(f"Request failed: {result}")
        else:
            processed_result = process_response(result)
            print(f"Processed result: {processed_result}")


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

以上架构通过各模块的协同工作，在异步编程的基础上，有效融合超时与重试机制，能够较好地应对高并发、不稳定网络连接等复杂情况。