asyncio 速度比 ThreadPool 慢？DeepL Pro 实测告诉你原因。

admin

```python
"""Translate using deeplpro's dl_session cookie."""
# pylint: disable=broad-exception-caught
import asyncio
import sys
from time import monotonic
from random import randrange
from typing import Union
from concurrent.futures import ThreadPoolExecutor
import httpx
from loadtext import loadtext
COOKIE = "dl_session=81f94b98-497f-4f18-ab79-1ef25076f4c9"
HEADERS = {"Content-Type": "application/json", "Cookie": COOKIE}
URL = "https://api.deepl.com/jsonrpc"
def deepl_tr(text: str, source_lang: str = "auto", target_lang: str = "zh") -> Union[dict, Exception]:
    """Translate using deeplpro's dl_session cookie."""
    data = {
        "jsonrpc": "2.0",
        "method": "LMT_handle_texts",
        "id": randrange(sys.maxsize),
        "params": {
            "splitting": "newlines",
            "lang": {"source_lang_user_selected": source_lang, "target_lang": target_lang},
            "texts": [{"text": text, "requestAlternatives": 3}],
        },
    }
    try:
        _ = httpx.post(URL, json=data, headers=HEADERS)
        jdata = _.json()
        return jdata
    except Exception as exc:
        return exc
async def deepl_tr_async(text: str, source_lang: str = "auto", target_lang: str = "zh") -> Union[dict, Exception]:
    """Translate using deeplpro's dl_session cookie."""
    data = {
        "jsonrpc": "2.0",
        "method": "LMT_handle_texts",
        "id": randrange(sys.maxsize),
        "params": {
            "splitting": "newlines",
            "lang": {"source_lang_user_selected": source_lang, "target_lang": target_lang},
            "texts": [{"text": text, "requestAlternatives": 3}],
        },
    }
    async with httpx.AsyncClient() as client:
        try:
            _ = await client.post(URL, json=data, headers=HEADERS)
            jdata = _.json()
            return jdata
        except Exception as exc:
            return exc
def main():
    """Run."""
    texts = loadtext(r"2024-08-20.txt")
    then = monotonic()
    with ThreadPoolExecutor() as executor:
        _ = executor.map(deepl_tr, texts)
    print(*_)
    time_el = monotonic() - then
    print(f"{len(texts)}, {time_el:.2f} {time_el / len(texts):.2f}")
async def main_a():
    """Run async."""
    texts = loadtext(r"2024-08-20.txt")
    then = monotonic()
    coros = [deepl_tr_async(text) for text in texts]
    _ = await asyncio.gather(*coros)
    print(_)
    time_el = monotonic() - then
    print(f"{len(texts)}, {time_el:.2f} {time_el / len(texts):.2f}")
if __name__ == "__main__":
    asyncio.run(main_a())
    main()
```