PDF翻译中的并发控制：用Semaphore防止API限流与资源耗尽-平芜编程栈

在批量翻译 PDF 的场景里，并发是最容易被低估的问题。一开始大家可能都会写一个简单的多线程循环，把几百个文件同时丢给翻译 API。结果往往有两种：要么触发对方的限流，大量请求失败；要么本地内存和连接数被占满，服务直接卡死。

这篇文章分享一个实用的并发控制方案：用 Python 的asyncio.Semaphore限制同时执行的翻译任务数，兼顾吞吐量和稳定性。

一、问题场景

假设你有一个翻译服务，需要批量处理 1000 个 PDF 文件。最朴素的写法可能是这样：

importconcurrent.futuresdeftranslate_file(path):# 调用翻译 APIreturnrequests.post(API_URL,files={"file":open(path,"rb")})withconcurrent.futures.ThreadPoolExecutor(max_workers=50)asexecutor:executor.map(translate_file,files)

这段代码在文件少的时候没问题，但一旦任务量变大，会遇到三类问题：

API 限流：翻译服务通常有 QPS 限制，超过后返回 429。
连接耗尽：大量并发连接占用本地端口和内存。
资源竞争：CPU/内存密集型任务（如 PDF 解析）同时运行，导致整体性能下降。

二、解决方案：Semaphore 限流

Semaphore是一个计数信号量，允许同时获取许可的协程数量有限。把它放在翻译任务入口，就能天然控制并发度。

importasyncioimportaiohttpimportaiofilesfrompathlibimportPath API_URL="https://api.example.com/translate"MAX_CONCURRENT=5# 根据 API 限流调整semaphore=asyncio.Semaphore(MAX_CONCURRENT)asyncdeftranslate_one(session:aiohttp.ClientSession,file_path:str)->dict:"""单个文件翻译，受 Semaphore 保护"""asyncwithsemaphore:# 同时最多 MAX_CONCURRENT 个协程进入asyncwithaiofiles.open(file_path,"rb")asf:data=awaitf.read()form=aiohttp.FormData()form.add_field("file",data,filename=Path(file_path).name)form.add_field("target_lang","zh")try:asyncwithsession.post(API_URL,data=form,timeout=30)asresp:resp.raise_for_status()result=awaitresp.json()return{"file":file_path,"status":"success","result":result}exceptasyncio.TimeoutError:return{"file":file_path,"status":"timeout"}exceptExceptionase:return{"file":file_path,"status":"error","message":str(e)}asyncdeftranslate_batch(file_paths:list[str])->list[dict]:"""批量翻译入口"""asyncwithaiohttp.ClientSession()assession:tasks=[translate_one(session,p)forpinfile_paths]returnawaitasyncio.gather(*tasks)

三、完整可运行示例

下面是一个带重试、进度日志和结果保存的完整示例：

importasyncioimportaiohttpimportaiofilesfrompathlibimportPathfromdatetimeimportdatetime API_URL="https://api.example.com/translate"MAX_CONCURRENT=5RETRY=2semaphore=asyncio.Semaphore(MAX_CONCURRENT)asyncdeftranslate_with_retry(session,file_path:str,retries:int=RETRY)->dict:"""带重试的单个文件翻译"""forattemptinrange(retries+1):result=awaittranslate_one(session,file_path)ifresult["status"]=="success"orattempt==retries:returnresultawaitasyncio.sleep(2**attempt)# 指数退避returnresultasyncdeftranslate_one(session:aiohttp.ClientSession,file_path:str)->dict:asyncwithsemaphore:try:asyncwithaiofiles.open(file_path,"rb")asf:data=awaitf.read()form=aiohttp.FormData()form.add_field("file",data,filename=Path(file_path).name)form.add_field("target_lang","zh")asyncwithsession.post(API_URL,data=form,timeout=30)asresp:ifresp.status==429:return{"file":file_path,"status":"rate_limited"}resp.raise_for_status()return{"file":file_path,"status":"success"}exceptasyncio.TimeoutError:return{"file":file_path,"status":"timeout"}exceptExceptionase:return{"file":file_path,"status":"error","message":str(e)}asyncdefmain():pdf_dir=Path("pdfs")files=[str(p)forpinpdf_dir.glob("*.pdf")]print(f"[{datetime.now()}] Start translating{len(files)}files, max_concurrent={MAX_CONCURRENT}")asyncwithaiohttp.ClientSession()assession:tasks=[translate_with_retry(session,f)forfinfiles]results=awaitasyncio.gather(*tasks)success=sum(1forrinresultsifr["status"]=="success")failed=len(results)-successprint(f"[{datetime.now()}] Done. success={success}, failed={failed}")# 保存失败列表，便于后续重试failed_files=[r["file"]forrinresultsifr["status"]!="success"]asyncwithaiofiles.open("failed_files.txt","w")asf:awaitf.write("\n".join(failed_files))if__name__=="__main__":asyncio.run(main())

四、Semaphore 与线程池的区别

很多人会问：用ThreadPoolExecutor(max_workers=N)不也能限流吗？

确实可以，但两者有本质区别：

维度	ThreadPoolExecutor	asyncio.Semaphore
并发模型	多线程，适合 CPU/IO 混合任务	单线程协程，适合高 IO 任务
资源占用	每个线程有独立栈空间，数量多时有内存压力	协程轻量，可创建成千上万个
适用场景	PDF 解析等 CPU 密集型操作	网络请求等 IO 密集型操作
灵活性	固定线程数，调整不够细粒度	可动态调整、可嵌套使用