本文介绍一种基于 asyncio 的轻量级、高并发、无预定义任务队列的持续请求方案，避免线程阻塞与响应缓存，支持动态调度、间隔控制与错误恢复，显著降低服务器负载。
本文介绍一种基于 asyncio 的轻量级、高并发、无预定义任务队列的持续请求方案，避免线程阻塞与响应缓存，支持动态调度、间隔控制与错误恢复，显著降低服务器负载。
在构建高频、长周期的网络爬取或服务探测系统时，传统 asyncio.gather() 要求预先构造完整任务列表，不仅内存开销大，还导致请求批次间存在明显延迟（如等待全部任务完成才发起下一轮）。而多线程方案（如示例中 threading.Thread + time.sleep）又易引发资源竞争、连接复用混乱及服务端过载。
理想的替代方案是：以协程为单元，按需、定时、无限循环地启动独立请求任务，不依赖全局任务池，且确保每个请求使用全新会话（禁用缓存、隔离 Cookie/Proxy）。以下是核心实现思路与可运行代码：
✅ 核心设计原则
- 无任务列表：用 asyncio.create_task() 动态创建任务，而非 gather() 预收集；
- 真异步间隔：使用 await asyncio.sleep() 替代 time.sleep()，避免阻塞事件循环；
- 请求隔离：每次请求均新建 aiohttp.ClientSession（或显式设置 cache=False, skip_auto_headers=['If-None-Match']）；
- 动态凭证/代理轮换：通过异步生成器（async def get_acc()）按需提供认证上下文；
- 错误韧性：结合 @retry_on_exception（适配协程版）与结构化异常捕获，防止单点失败中断整个流程。
? 示例代码（asyncio 重构版）
import asyncio
import aiohttp
import orjson
from loguru import logger
from auth import Auth # 假设已适配 async 方法（如 get_headers_async）
# 全局配置
URL = "https://example.com/api"
CONST1, CONST2 = "...", "..."
# 异步凭证轮换生成器
async def get_acc():
acc_paths = ["accounts/acc1.json", "accounts/acc2.json"]
accs = [Auth(path) for path in acc_paths]
while True:
for acc in accs:
yield acc
await asyncio.sleep(0.01) # 防止单一账号过载
# 异步请求函数（带重试）
async def func1(session: aiohttp.ClientSession, acc, info):
data = {"key": CONST1, "info": info}
headers = await acc.get_headers_async() # 假设已异步化
cookies = await acc.get_cookies_async()
async with session.post(
URL,
headers=headers,
cookies=cookies,
json=data,
timeout=aiohttp.ClientTimeout(total=15),
) as resp:
assert resp.status == 200
return await resp.json()
async def func2(session: aiohttp.ClientSession, acc, info1, info2):
data = {"key": CONST2, "a": info1, "b": info2}
headers = await acc.get_headers_async()
cookies = await acc.get_cookies_async()
async with session.post(URL, headers=headers, cookies=cookies, json=data) as resp:
if resp.status != 200:
logger.warning("func2 failed: {}", resp.status)
return
result = await resp.json()
if result.get("code") == "ERROR" and result.get("error"):
logger.info("Remote error: {}", result["error"])
return
info = result.get("payload")
await func1(session, acc, info)
# 主工作协程（每轮请求独立 session，彻底隔离）
async def worker(proxy: dict = None):
acc_gen = get_acc()
while True:
try:
acc = await anext(acc_gen)
# 每次请求使用全新 session → 确保无缓存、无连接复用污染
timeout = aiohttp.ClientTimeout(total=20)
connector = aiohttp.TCPConnector(limit=100, limit_per_host=30)
async with aiohttp.ClientSession(
connector=connector,
timeout=timeout,
# 关键：禁用自动缓存头 & 强制刷新
skip_auto_headers={"If-None-Match", "If-Modified-Since"},
) as session:
async with session.get(
URL,
cookies=await acc.get_cookies_async(),
proxy=proxy.get("http") if proxy else None,
) as resp:
if resp.status != 200:
continue
data = orjson.loads(await resp.text())
for item in data.get("items", []):
info1, info2, info3 = item["a"], item["b"], item["c"]
# 启动独立任务，不阻塞主循环
asyncio.create_task(
func2(session, acc, info1, info2)
)
except Exception as e:
logger.exception("Worker error: {}", e)
finally:
await asyncio.sleep(0.04) # 精确控制请求频率（40ms/次）
# 启动入口
async def main():
proxy_list = [None, {"http": "http://user:pass@proxy:8080"}]
# 并发运行多个 worker（对应不同代理/策略）
tasks = [asyncio.create_task(worker(proxy)) for proxy in proxy_list]
await asyncio.gather(*tasks, return_exceptions=True)
if __name__ == "__main__":
logger.add("logs.log", level="DEBUG", filter=lambda r: r["extra"].get("logs"))
asyncio.run(main())
⚠️ 关键注意事项
- Cookie/Headers 异步化：原 Auth 类需补充 async 方法（如 get_headers_async()），否则会阻塞事件循环；
- 连接池调优：TCPConnector(limit_per_host=30) 可防止单域名连接风暴；根据目标服务 QPS 合理设置 limit；
- 内存安全：create_task() 启动的任务需自行处理生命周期，避免无限堆积（建议在 func2 内部加 try/finally 或超时控制）；
- 日志性能：高频场景下禁用 logger.debug(..., resp_text=...)（序列化耗时），改用结构化字段或采样打印；
- 反爬适配：若目标服务校验 User-Agent 或 TLS 指纹，需在 ClientSession 中注入 Fingerprint 或使用 curl_cffi 等库。
该方案将请求节奏从「批处理」转为「流式发射」，既保持低延迟、高吞吐，又完全规避了线程模型的资源争用与同步开销，是现代异步 IO 场景下的推荐实践。
