Python异步函数立即返回调度及代理缓存无阻塞实现方案

嘿，这个需求在异步服务开发里挺常见的，咱们一步步拆解实现，同时把你关心的「异步函数调度立即返回」的问题讲清楚。

你要的效果是客户端请求时不阻塞，立即返回提示，后台异步完成API调用和缓存写入。在asyncio生态里，关键就是不要直接await你的fetch函数，而是把它包装成一个后台任务，丢给事件循环去执行，处理客户端请求的协程直接继续走返回逻辑。

我们用aiohttp作为web框架（和你用的请求库一致），同时处理缓存的并发安全问题，避免多请求冲突。

1. 基础依赖与缓存初始化 #

先导入需要的库，定义缓存和异步锁（异步环境下多个协程操作同一个字典会有竞态风险）：

import asyncio
import aiohttp
from datetime import datetime
from aiohttp import web

# 缓存存储：key是页码，value是(响应数据, 缓存时间)
cache = {}
# 异步锁，保护缓存的读写操作
cache_lock = asyncio.Lock()

2. 优化异步请求函数 #

给你的fetch函数加上缓存锁，保证写入缓存时的原子性，同时处理异常情况（比如API请求失败）：

async def fetch(url, page_num):
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                # 确保API返回成功，再处理数据
                response.raise_for_status()
                resp = await response.json()
                # 用锁保护缓存写入，避免并发冲突
                async with cache_lock:
                    cache[page_num] = (resp, datetime.now())
                return resp
    except Exception as e:
        # 请求失败时，清理缓存里的"正在加载"标记
        async with cache_lock:
            if page_num in cache:
                del cache[page_num]
        print(f"Failed to fetch page {page_num}: {str(e)}")

3. 处理客户端请求的核心逻辑 #

这里是关键：当缓存不存在时，用asyncio.create_task()把fetch丢到后台执行，立即返回提示。同时添加「正在加载」标记，避免同一页码被重复请求API：

async def proxy_handler(request):
    # 从请求参数里获取页码，默认1
    page_num = request.query.get('page', '1')
    
    async with cache_lock:
        # 检查缓存状态
        if page_num in cache:
            cache_value = cache[page_num]
            if cache_value is not None:
                # 缓存已就绪，直接返回数据
                resp_data, _ = cache_value
                return web.json_response(resp_data)
            else:
                # 正在加载中，直接返回提示
                return web.json_response({"message": "we're working on it"})
        
        # 标记为正在加载，避免重复请求
        cache[page_num] = None
    
    # 把fetch任务提交给事件循环，后台执行，不等待完成
    target_api_url = f"https://your-target-api.com/data?page={page_num}"  # 替换成实际API地址
    asyncio.create_task(fetch(target_api_url, page_num))
    
    # 立即返回提示给客户端
    return web.json_response({"message": "we're working on it"})

4. 启动代理服务器 #

最后把handler注册到aiohttp web应用，启动服务：

if __name__ == "__main__":
    app = web.Application()
    app.add_routes([web.get('/get-data', proxy_handler)])
    # 启动服务，监听本地8080端口
    web.run_app(app, host='0.0.0.0', port=8080)

• asyncio.create_task()：这是实现「调度异步函数立即返回」的核心。它会把你的协程包装成一个Task对象，注册到当前的事件循环中，事件循环会在空闲时自动执行这个任务。调用create_task()后，当前协程不会等待任务完成，而是直接继续执行后面的代码——这就是你要的「不阻塞客户端，立即返回」的效果。
• 异步锁asyncio.Lock：因为多个客户端请求可能同时访问缓存，比如两个请求同时发现某页码不在缓存里，如果不加锁，会同时发起两次API请求。用锁可以保证缓存的读写操作是原子性的，避免竞态条件。
• 「正在加载」标记：在缓存里把页码设为None，后续请求进来就知道这个数据正在加载，不会重复触发API请求，减少资源浪费。

你可以用curl或者浏览器访问http://localhost:8080/get-data?page=1，第一次访问会立即返回{"message": "we're working on it"}，等几秒后再访问同一页码，就能拿到缓存好的API数据了。

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