连接

服务器代理

方案 B 使用
LR232、WECHAT 此类配置 ip 白名单的平台需要使用代理连接(本地服务无固定 ip，采用服务器 ip 作为白名单)
command 容器中将 0.0.0.0:5923 端口使用 sock5 转发到服务器上，故代理连接需要连接至 command:5923 来实现转发
使用 connect(True) 可实现代理转发，使用 connect() 可实现本机正常访问
由于 request 是阻塞的，所以返回 httpx 对象
采用上下文来关闭当前连接
使用示例:

async with connect() as client:
    response = await client.get(url, headers=headers)
    if response.status_code == 200:
    else:

mihomo 代理

方案 A 使用
服务器上无代理环境，瓦片地图等资源经常加载不出来
使用 sock 方法连接至 mihomo 的 7891 端口

connect 函数

针对方案 A 和方案 B 进行兼容设计
需要用到服务器代理的使用connect(True)即可，会自动判断有无服务器环境
- 有服务器配置-> 方案 B 环境，转发 command:5923
- 无服务器配置-> 方案 A 环境，本身在服务器，直接请求
需要用 mihomo 转发时使用connect(use_agent=True)即可
- 能检测到端口即转发
- 此项会覆盖前一项，即二者都设置成 Ture 仍会使用 mihomo 转发

连接关闭

若连接未关闭，在多次调用后电脑网络适配器可能卡死
以下代码展示了关闭和未关闭两种情况

示例代码

import os
import httpx
import psutil
import asyncio
from contextlib import asynccontextmanager

# 方法1：改造后的 connect
@asynccontextmanager
async def connect():
    client = httpx.AsyncClient()
    try:
        yield client
    finally:
        await client.aclose()


# 打印当前进程的 TCP 连接数
def print_connections(tag=""):
    pid = os.getpid()
    p = psutil.Process(pid)
    conns = p.connections(kind="tcp")
    print(f"[{tag}] 当前 TCP 连接数: {len(conns)}")


# 测试1：不带 close
async def test_no_close():
    print("\n=== 测试: 没有 close (直接用 AsyncClient) ===")
    client = httpx.AsyncClient()
    r = await client.get("https://bilibili.com")
    print("请求完成:", r.status_code)
    print_connections("请求后 (no close)")

    # 不调用 client.aclose()
    del client
    await asyncio.sleep(1)  # 给 GC 一点时间
    print_connections("GC 后 (no close)")


# 测试2：带 close
async def test_with_close():
    print("\n=== 测试: 有 close (使用 async with connect) ===")
    async with connect() as client:
        r = await client.get("https://bilibili.com")
        print("请求完成:", r.status_code)
        print_connections("请求后 (with close)")

    # 出了 async with 块，client 已关闭
    await asyncio.sleep(1)
    print_connections("GC 后 (with close)")


# ========== 主函数 ==========
async def main():
    await test_no_close()
    await asyncio.sleep(1)
    await test_with_close()


if __name__ == "__main__":
    asyncio.run(main())

其他方法

以下展示了修改代理的两种旧方法

方法1:修改全局配置（仅适合全局都需要代理的情况）

import os
import asyncio
import aiohttp
import requests
from flask import Flask
from ssl import SSLContext
from aiohttp import ClientSession


# 设置SOCKS5代理
os.environ["HTTP_PROXY"] = "socks5://127.0.0.1:1080"
os.environ["HTTPS_PROXY"] = "socks5://127.0.0.1:1080"

response = requests.get("http://ifconfig.me")
print(response.text)

方法2:传入自定义connector(旧方法不如connect简便)

import aiohttp
import asyncio
from aiohttp_socks import ProxyConnector


class CustomClientSession(aiohttp.ClientSession):
    def __init__(self, *args, connector=None, **kwargs):
        # SOCKS5 代理配置
        proxy_host = "127.0.0.1"  # SOCKS5 代理地址
        proxy_port = 1080  # SOCKS5 代理端口

        # 如果传入了connector，获取connector中的配置
        ssl_context = getattr(
            connector, "ssl", SSLContext()
        )  # 获取SSL上下文，默认为默认SSLContext
        limit_per_host = getattr(
            connector, "limit_per_host", 500
        )  # 获取连接限制，默认为500
        force_close = getattr(
            connector, "force_close", True
        )  # 获取force_close，默认为True

        # 创建 ProxyConnector，使用传入的connector中的参数
        connector = ProxyConnector(
            host=proxy_host,
            port=proxy_port,
        )

        # 在初始化时设置代理连接器
        super().__init__(*args, connector=connector, **kwargs)


# 使用自定义的 ClientSession 替换默认实现
aiohttp.ClientSession = CustomClientSession


async def fetch(url):
    from aiohttp import WSMessage, ClientWebSocketResponse, TCPConnector

    connector = TCPConnector(limit=10, ssl=SSLContext())
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get(url) as response:
            return await response.text()


async def main():
    url = "http://ifconfig.me"
    content = await fetch(url)
    print(content)


# 运行测试
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

服务器代理​

mihomo 代理​

connect 函数​

连接关闭​

相关知识​

其他方法​

服务器代理

mihomo 代理

connect 函数

连接关闭

相关知识

其他方法