本文介绍如何摆脱 asyncio.gather() 对预定义任务列表的依赖，构建一个可长期运行、低延迟、无缓存、支持动态请求生成的异步 HTTP 请求调度器，兼顾服务器负载控制与会话隔离。
本文介绍如何摆脱 `asyncio.gather()` 对预定义任务列表的依赖，构建一个可长期运行、低延迟、无缓存、支持动态请求生成的异步 http 请求调度器，兼顾服务器负载控制与会话隔离。
在实际爬虫或服务轮询场景中，常见需求是“持续按固定间隔发起请求”，但传统 asyncio.gather(tasks) 要求一次性构造全部协程对象——这不仅违背流式调度逻辑，还会造成内存累积、响应延迟（如每轮 gather 执行完才启动下一轮），且难以实现请求节流、会话轮换与缓存规避。
正确的解法是：用 asyncio.create_task() 动态创建并管理任务，配合 asyncio.sleep() 精确控制间隔，结合 aiohttp 实现真正非阻塞、会话隔离、无共享缓存的异步请求流。
以下是一个生产就绪的示例架构：
import asyncio
import aiohttp
import time
from loguru import logger
from typing import AsyncIterator, Dict, Any
# 全局配置（避免重复加载）
URL = "https://api.example.com/endpoint"
REQUEST_INTERVAL = 0.04 # 秒级间隔，可动态调整
MAX_CONCURRENT = 5 # 并发上限，防压垮服务
class AuthSession:
"""轻量会话封装，确保每次请求使用独立 headers/cookies"""
def __init__(self, auth_path: str):
self.auth_path = auth_path
# 实际中从文件/DB加载认证凭据，不复用 session 对象
def get_headers(self) -> Dict[str, str]:
return {"User-Agent": "AsyncBot/1.0", "X-Request-ID": f"{int(time.time() * 1000)}"}
def get_cookies(self) -> Dict[str, str]:
return {"session_id": f"sess_{hash(self.auth_path) % 1000000}"}
async def fetch_with_retry(
session: aiohttp.ClientSession,
url: str,
method: str = "GET",
headers: Dict[str, str] = None,
cookies: Dict[str, str] = None,
json: Dict[str, Any] = None,
max_retries: int = 3
) -> aiohttp.ClientResponse:
for attempt in range(max_retries):
try:
async with session.request(
method=method,
url=url,
headers=headers,
cookies=cookies,
json=json,
# 关键：禁用客户端缓存
params={"t": str(int(time.time() * 1000))}, # 时间戳参数强制刷新
timeout=aiohttp.ClientTimeout(total=15)
) as resp:
resp.raise_for_status()
return resp
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
if attempt == max_retries - 1:
logger.error("Request failed after {n} retries: {url}", n=max_retries, url=url)
raise e
await asyncio.sleep(1 * (2 ** attempt)) # 指数退避
raise RuntimeError("Unreachable")
async def request_worker(
worker_id: int,
auth_gen: AsyncIterator[AuthSession],
proxy: Dict[str, str] = None
):
"""单个持续工作的协程：无限循环 + 动态会话 + 精确间隔"""
connector = aiohttp.TCPConnector(
limit_per_host=MAX_CONCURRENT,
ttl_dns_cache=300,
keepalive_timeout=15
)
async with aiohttp.ClientSession(connector=connector) as session:
while True:
try:
# 动态获取会话（模拟轮换账号）
acc = await anext(auth_gen)
headers = acc.get_headers()
cookies = acc.get_cookies()
# 第一次请求：获取基础数据
resp1 = await fetch_with_retry(
session, URL + "/list", "GET", headers, cookies
)
data = await resp1.json()
# 解析并触发后续请求（如需）
for item in data.get("items", [])[:3]: # 限制每轮请求数
info1 = item.get("id")
info2 = item.get("type")
# 避免重复处理（可用 Redis 或内存 set 去重）
if not hasattr(request_worker, "_seen"):
request_worker._seen = set()
if info1 in request_worker._seen:
continue
request_worker._seen.add(info1)
# 异步触发二级请求（不等待，fire-and-forget）
asyncio.create_task(
secondary_request(session, acc, info1, info2)
)
logger.debug("Worker {id} completed cycle", id=worker_id)
except Exception as e:
logger.exception("Worker {id} error: {e}", id=worker_id, e=e)
# 严格间隔：sleep 到下一个调度点（补偿执行耗时）
await asyncio.sleep(REQUEST_INTERVAL)
async def secondary_request(
session: aiohttp.ClientSession,
acc: AuthSession,
info1: str,
info2: str
):
"""辅助请求，完全异步执行，不影响主循环节奏"""
try:
payload = {"ref_id": info1, "category": info2}
resp = await fetch_with_retry(
session, URL + "/detail", "POST", acc.get_headers(), acc.get_cookies(), json=payload
)
result = await resp.json()
logger.info("Detail fetched for {id}: {code}", id=info1, code=result.get("code"))
except Exception as e:
logger.warning("Secondary request failed for {id}: {e}", id=info1, e=e)
# 会话生成器（支持无限、轮换、懒加载）
async def auth_session_generator() -> AsyncIterator[AuthSession]:
auth_paths = ["accounts/acc1.json", "accounts/acc2.json"]
while True:
for path in auth_paths:
yield AuthSession(path)
await asyncio.sleep(0) # 让出控制权，避免饥饿
# 主调度入口
async def main():
logger.info("Starting async request scheduler...")
# 启动多个独立 worker（模拟多代理/多账号并发）
proxy_list = [None] # 可扩展为真实代理池
workers = []
for i, proxy in enumerate(proxy_list):
auth_gen = auth_session_generator()
task = asyncio.create_task(
request_worker(worker_id=i, auth_gen=auth_gen, proxy=proxy)
)
workers.append(task)
# 保持主协程活跃（可加入信号监听、热重载等）
try:
await asyncio.gather(*workers)
except KeyboardInterrupt:
logger.info("Shutting down gracefully...")
for task in workers:
task.cancel()
await asyncio.gather(*workers, return_exceptions=True)
if __name__ == "__main__":
asyncio.run(main())✅ 关键优势说明：
- 无预定义任务：create_task() 动态创建，while True 实现无限流式调度；
- 低延迟响应：每轮仅 sleep(0.04)，不等待整批完成，新请求随时插入；
- 零缓存干扰：通过时间戳参数 ?t=... + aiohttp 默认禁用缓存 + 独立 ClientSession；
- 负载可控：TCPConnector.limit_per_host 限流 + REQUEST_INTERVAL 全局节制 + 指数退避容错；
- 会话隔离：每个请求使用全新 headers/cookies，无跨请求状态污染；
- 可观测性强：结构化日志、错误隔离、worker 级别监控。
⚠️ 注意事项：
- 避免在 request_worker 中使用 time.sleep()（会阻塞整个事件循环），务必用 await asyncio.sleep()；
- 内存去重（_seen）仅适用于短期轻量场景；高吞吐建议接入 Redis 的 SETNX 或布隆过滤器；
- 若目标接口有反爬，请将 User-Agent、Referer、请求头指纹等纳入 AuthSession 动态生成；
- 生产环境应添加 asyncio.CancelledError 处理，并在退出前关闭所有连接器。
该模式已广泛用于实时行情拉取、分布式任务分发、API 健康巡检等场景，兼具弹性、可观测性与工程健壮性。
