使用 asyncio 实现无固定任务列表的持续异步请求调度-Python教程-PHP中文网

本文介绍如何摆脱 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())

✅ 关键优势说明：