代码干净清爽才能看着赏心悦目：
#!/usr/bin/env python3.11
import time
from contextlib import contextmanager
from enum import StrEnum
import anyio
import httpx
@contextmanager
def timeit(msg: str):
start = time.time()
yield
cost = time.time() - start
print(msg, f"{cost = }")
class CdnHost(StrEnum):
jsdelivr = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5.9.0/swagger-ui.css"
unpkg = "https://unpkg.com/swagger-ui-dist@5.9.0/swagger-ui.css"
cloudflare = (
"https://cdnjs.cloudflare.com/ajax/libs/swagger-ui/5.9.0/swagger-ui.css"
)
async def find_fastest_host(loop_interval=0.1) -> str:
urls = list(CdnHost)
results = [None] * len(urls)
async def fetch(httpx_client, url, index):
try:
r = await httpx_client.get(url)
except (httpx.ConnectError, httpx.ReadError):
...
else:
print(f"{url = }\n{r.elapsed = }")
if r.status_code < 300:
results[index] = r.content
async with (anyio.create_task_group() as tg, httpx.AsyncClient() as client):
for i, url in enumerate(urls):
tg.start_soon(fetch, client, url, i)
for _ in range(int(5 / loop_interval) + 1):
if any(r is not None for r in results):
tg.cancel_scope.cancel()
break
await anyio.sleep(0.1)
for url, res in zip(urls, results):
if res is not None:
return url
return urls[0]
async def main():
with timeit("Sniff hosts"):
url = await find_fastest_host()
print("result:", url)
if __name__ == "__main__":
anyio.run(main)
执行结果：
url = <CdnHost.unpkg: 'https://unpkg.com/swagger-ui-dist@5.9.0/swagger-ui.css'>
r.elapsed = datetime.timedelta(seconds=1, microseconds=418428)
Sniff hosts cost = 1.4775612354278564
result: https://unpkg.com/swagger-ui-dist@5.9.0/swagger-ui.css