使用Python协程并发测试cdn响应速度

代码干净清爽才能看着赏心悦目：

#!/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/[email protected]/swagger-ui.css"
    unpkg = "https://unpkg.com/[email protected]/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/[email protected]/swagger-ui.css'>
r.elapsed = datetime.timedelta(seconds=1, microseconds=418428)
Sniff hosts cost = 1.4775612354278564
result: https://unpkg.com/[email protected]/swagger-ui.css