Python 異步並發提示詞：asyncio 批量請求改造（限制並發數、超時、重試、結果按順序收集）
腳本要調用幾千次接口或下載大量文件、一個個串行跑太慢時用：把同步代碼交給 AI，改寫成 asyncio 並發版本，帶並發上限、超時、失敗重試、進度顯示和錯誤彙總，並說明什麼時候其實用線程池更合適。
通用大模型 對話模型通用
你是一名熟悉 Python asyncio 的工程師。請把下面的串行代碼改成並發執行。 - Python 版本：[如 3.11] - 現有代碼（串行版本）： [粘貼代碼] - 任務規模：[如 5000 個 URL] - 對方服務的限制：[如每秒最多 20 次請求，或未知] - 失敗處理要求：[如失敗重試 3 次，最終失敗的記錄到文件] - 能否換用異步 HTTP 庫：[可以/只能用 requests] 要求： 1. 先判斷：這個任務是等待網絡或磁盤為主（適合 asyncio 或線程池），還是計算為主（應該用多進程）。如果不能換用異步庫，說明用線程池的寫法。 2. 並發控制：用信號量限制同時進行的請求數；對方有速率限制時，額外做速率控制，並說明兩者的區別。 3. 連接複用：整個任務共用一個會話或客戶端對象，不要每個請求新建一個。 4. 超時：每個請求有超時；說明整體超時的設定方式。 5. 重試：只對可重試的錯誤（超時、連接錯誤、5xx、429）重試，指數退避；不可重試的錯誤直接記錄。 6. 結果收集：結果與輸入一一對應（保持順序或帶上輸入標識）；單個任務失敗不影響其他任務；最後彙總成功數、失敗數和失敗原因分佈。 7. 進度：顯示已完成數量。 8. 正確使用 asyncio：入口用標準的運行方式；說明在 Jupyter 中運行時的區別；不要在協程中調用阻塞函數（如 time.sleep、同步的文件讀寫大文件），需要時放到線程中執行。 只針對我有權訪問的接口，遵守對方的速率限制和服務條款。輸出：完整代碼（中文註釋）、依賴安裝命令、參數怎麼調。
高亮處換成你自己的內容：[如 3.11]、[粘貼代碼]、[如 5000 個 URL]、[如每秒最多 20 次請求，或未知]、[如失敗重試 3 次，最終失敗的記錄到文件]、[可以/只能用 requests]
ChatGPT Plus 充值
已被複製 0 次
使用說明
怎麼填變量：[對方服務的限制] 一定要尊重。並發數不是越大越好：超過對方的限制會被封禁或返回大量 429，反而更慢。不知道限制時，從較小的並發（例如 5 到 10）開始逐步調。[能否換用異步庫] 決定了方案：requests 是同步庫，放在協程裏直接調用會阻塞，要麼換異步 HTTP 庫，要麼用線程池。
常見坑：
- 在 async 函數里調用 requests 或 time.sleep，代碼看起來是異步的，實際仍是一個個串行執行。
- 一次性為幾千個任務創建協程並同時發出，沒有並發上限，瞬間打滿對方服務或耗盡本機的連接和文件描述符。
- 用 gather 收集結果時沒有處理異常，一個任務出錯就導致整體失敗或其他結果丟失。可以開啟返回異常的選項，或在每個任務內部捕獲。
追問技巧：追問「把結果邊完成邊寫入文件，中途中斷後可以從斷點繼續，不重複請求已完成的部分」。
示例輸出
示例，僅供參考（使用 httpx 異步客戶端，節選）
python
import asyncio
import httpx
CONCURRENCY = 10
RETRIABLE = {429, 500, 502, 503, 504}
async def fetch(client: httpx.AsyncClient, sem: asyncio.Semaphore, url: str) -> tuple[str, int | str]:
async with sem:
for attempt in range(3):
try:
r = await client.get(url, timeout=10)
if r.status_code in RETRIABLE:
raise httpx.HTTPStatusError("可重試", request=r.request, response=r)
return url, r.status_code
except (httpx.TimeoutException, httpx.TransportError, httpx.HTTPStatusError) as e:
if attempt == 2:
return url, f"失敗：{type(e).__name__}"
await asyncio.sleep(2 ** attempt)
async def main(urls: list[str]) -> None:
sem = asyncio.Semaphore(CONCURRENCY)
async with httpx.AsyncClient() as client:
results = await asyncio.gather(*(fetch(client, sem, u) for u in urls))
failed = [r for r in results if isinstance(r[1], str)]
print(f"完成 {len(results)}，失敗 {len(failed)}")
asyncio.run(main(urls))同款作品
用這條提示詞做出來的作品；原作者會因此獲得積分
還沒有同款，來做第一個。
0 條評論
還沒有評論，來搶沙發～