创建新aiohttp.ClientSession仍报'Session is closed'错误求助

问题根源 #

你遇到的Session is closed错误，核心原因是全局定义的TCPConnector被多个ClientSession共享，第一个session关闭时会连带关闭这个connector，后续新建的session使用已关闭的connector就会触发错误。

看你的代码，connector = aiohttp.TCPConnector(ssl=ssl_context)是全局变量，每次调用compile_calls_get且ssl_verify=True时，都会把这个全局connector传给新的ClientSession。而async with aiohttp_session as session:结束时，session会自动关闭，同时会关闭它使用的connector（默认share_connector=False，即session独占connector）。所以第一次请求后，这个全局connector已经被关闭，第二次重试时新建的session用这个connector，自然就报session closed的错。

解决方案 #

方案1：每次创建session时新建专属connector #

把全局的connector移到compile_calls_get函数内部，每次新建session时都创建对应的connector，避免复用已关闭的实例：

async def compile_calls_get(url_set: list,
                            headers: Optional[dict] = None,
                            ssl_verify: bool = False) -> tuple:
    if ssl_verify:
        ssl_context = ssl.create_default_context(cafile=creds.PROXY_PATH)
        connector = aiohttp.TCPConnector(ssl=ssl_context)
        async with aiohttp.ClientSession(headers=headers, connector=connector) as session:
            extracted_data = await asyncio.gather(*[fetch_get(url=url, session=session) for url in url_set])
        return extracted_data
    else:
        connector = aiohttp.TCPConnector(ssl=False)
        async with aiohttp.ClientSession(headers=headers, connector=connector) as session:
            extracted_data = await asyncio.gather(*[fetch_get(url=url, session=session) for url in url_set])
        return extracted_data

这样每次请求都会创建全新的connector和session，互相独立，不会出现之前的关闭冲突。

方案2：让connector被多个session共享（设置share_connector=True） #

如果想复用connector来提升性能，可以在创建ClientSession时指定share_connector=True，这样session关闭时不会关闭connector，直到所有使用它的session都关闭后再手动关闭：

# 保留全局connector
ssl_context = ssl.create_default_context(cafile=creds.PROXY_PATH)
connector = aiohttp.TCPConnector(ssl=ssl_context)
proxy = "example_proxy"

async def compile_calls_get(url_set: list,
                            headers: Optional[dict] = None,
                            ssl_verify: bool = False) -> tuple:
    if ssl_verify:
        async with aiohttp.ClientSession(headers=headers, connector=connector, share_connector=True) as session:
            extracted_data = await asyncio.gather(*[fetch_get(url=url, session=session) for url in url_set])
        return extracted_data
    else:
        connector_no_ssl = aiohttp.TCPConnector(ssl=False)
        async with aiohttp.ClientSession(headers=headers, connector=connector_no_ssl, share_connector=True) as session:
            extracted_data = await asyncio.gather(*[fetch_get(url=url, session=session) for url in url_set])
        return extracted_data

# 注意：程序结束时需要手动关闭全局connector
# 比如在extract_data_asynchronously最后或者调用结束后加：
# asyncio.run(connector.close())

不过这种方式需要注意在程序生命周期结束时手动关闭connector，避免资源泄漏。

额外优化：给请求添加内置重试逻辑 #

现在的重试是把所有失败的URL收集后重新批量请求，其实可以在fetch_get函数内部直接添加重试逻辑，不需要外层循环，代码更简洁：

async def fetch_get(url: str, session: aiohttp.ClientSession, max_retries: int = 3) -> Any:
    for attempt in range(max_retries + 1):
        try:
            async with session.get(url=url, proxy=proxy) as response:
                if 500 <= response.status < 600:
                    if attempt == max_retries:
                        return {"success": response.status, "url": url, "data": None}
                    await asyncio.sleep(2 ** attempt)  # 指数退避，避免频繁请求
                    continue
                data = await response.text()
                return {"success": response.status, "url": url, "data": data}
        except aiohttp.ClientError as e:
            if attempt == max_retries:
                return {"success": -1, "url": url, "data": str(e)}
            await asyncio.sleep(2 ** attempt)

这样每个请求会自动重试指定次数，外层只需要一次批量请求即可，减少循环逻辑。

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