基于aiohttp整合HTTP客户端与服务端功能的开发方案咨询

核心问题本质 #

• loop.run_until_complete() 与 web.run_app() 均为绑定独立事件循环的阻塞入口，同一线程中无法同时并行启动两个阻塞循环，这是之前无法同时跑通两端逻辑的根本原因。
• 参考的 AppRunner 启动逻辑本身没有问题，所有await调用必须处于同一个异步上下文内，在普通同步函数中直接编写await语句是不符合Python异步语法规则的，这是之前代码无法运行的核心原因。

常见认知误区：将服务端启动逻辑与客户端拉取逻辑拆到两个独立的同步入口中，分别调用阻塞方法启动，最终必然出现两个循环互相抢占控制权的问题。

标准实现结构 #

所有异步逻辑统一挂载到同一个事件循环中运行，不需要拆分多个阻塞入口，整体流程分为四个部分：

1. 定义跨协程共享的资源存储，用于存放客户端拉取、处理后的最终数据
2. 编写后台常驻协程：定期从外网拉取原始数据、完成格式重排与封装、将处理后的结果更新到共享存储
3. 编写服务端请求处理协程：从共享存储读取最新数据，封装为HTTP响应返回给调用方
4. 在单个异步主函数中，并行完成三个动作：通过AppRunner启动服务端监听、注册后台数据拉取任务、挂起事件循环避免程序退出

可直接运行的示例代码 #

import aiohttp
import asyncio
from aiohttp import web

# 共享数据存储，实际生产中建议绑定到app对象上而非全局变量
shared_data = None
# 初始化异步锁，用于保证数据更新的原子性
data_lock = asyncio.Lock()

async def periodic_fetch_task():
    """后台常驻协程：定期拉取外网数据并处理"""
    global shared_data
    remote_fetch_url = "替换为实际需要拉取的外网接口地址"
    while True:
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(remote_fetch_url, timeout=10) as resp:
                    raw_resp = await resp.json()
            # 自定义格式重排、封装逻辑
            processed_data = {
                "last_refresh_timestamp": asyncio.get_running_loop().time(),
                "data_payload": [
                    {
                        "field_a": item.get("source_field_a"),
                        "field_b": item.get("source_field_b")
                    } for item in raw_resp.get("list", [])
                ]
            }
            # 加锁完成数据更新
            async with data_lock:
                shared_data = processed_data
        except Exception as e:
            print(f"数据拉取流程异常: {str(e)}")
        # 每30秒刷新一次
        await asyncio.sleep(30)

async def data_api_handler(request):
    """服务端接口处理协程：对外提供最新的处理后数据"""
    global shared_data
    async with data_lock:
        return web.json_response(
            shared_data or {"msg": "数据尚未完成首次初始化"}
        )

async def main():
    """统一异步入口：所有逻辑都在同一个事件循环中运行"""
    # 初始化服务端应用
    app = web.Application()
    app.add_routes([web.get("/api/latest", data_api_handler)])
    
    # 非阻塞启动服务端
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "0.0.0.0", 8080)
    await site.start()
    print("HTTP服务端已启动，监听8080端口")

    # 并行启动后台数据拉取任务
    asyncio.create_task(periodic_fetch_task())

    # 永久挂起事件循环，避免程序直接退出
    await asyncio.Event().wait()

if __name__ == "__main__":
    # 仅启动一次事件循环，所有异步逻辑均在此循环内调度
    asyncio.run(main())

数据共享与锁的使用规则 #

• aiohttp基于单线程事件循环运行，大部分场景不存在多线程读写竞态问题，是否需要加锁取决于数据更新的流程是否存在原子性要求：
  • 若数据更新为单步原子赋值（如直接将完整处理好的字典赋值给存储变量，赋值过程中没有await等待），则完全不需要加锁，单线程调度下不会出现读写半截数据的问题。
  • 若数据更新涉及多步await等待（如先读取旧值、再发起请求拉取新值、最后合并新旧值写回存储），则必须使用asyncio.Lock()对更新、读取流程加锁，避免协程切换过程中出现脏数据。
• 生产环境建议将共享存储直接绑定到aiohttp的app对象上（如app["shared_storage"] = {}），在接口处理函数中可以通过request.app直接访问，避免全局变量的作用域问题，也兼容后续多worker部署的场景。

常见踩坑点 #

• 不要在代码中同时混用web.run_app()与手动创建事件循环的调用逻辑，两者都会自动接管事件循环控制权，必然出现互斥阻塞问题。
• 所有涉及await的异步逻辑必须定义为async修饰的协程函数，同步函数中无法直接编写await语句，所有异步逻辑需要通过asyncio.run()统一挂载到入口协程上。
• 后台拉取任务建议添加异常捕获与超时设置，避免单次请求失败导致整个后台协程崩溃退出。

内容的提问来源于stack exchange，提问作者stacker