在批量翻译 PDF 的场景里并发是最容易被低估的问题。一开始大家可能都会写一个简单的多线程循环把几百个文件同时丢给翻译 API。结果往往有两种要么触发对方的限流大量请求失败要么本地内存和连接数被占满服务直接卡死。这篇文章分享一个实用的并发控制方案用 Python 的asyncio.Semaphore限制同时执行的翻译任务数兼顾吞吐量和稳定性。一、问题场景假设你有一个翻译服务需要批量处理 1000 个 PDF 文件。最朴素的写法可能是这样importconcurrent.futuresdeftranslate_file(path):# 调用翻译 APIreturnrequests.post(API_URL,files{file:open(path,rb)})withconcurrent.futures.ThreadPoolExecutor(max_workers50)asexecutor:executor.map(translate_file,files)这段代码在文件少的时候没问题但一旦任务量变大会遇到三类问题API 限流翻译服务通常有 QPS 限制超过后返回 429。连接耗尽大量并发连接占用本地端口和内存。资源竞争CPU/内存密集型任务如 PDF 解析同时运行导致整体性能下降。二、解决方案Semaphore 限流Semaphore是一个计数信号量允许同时获取许可的协程数量有限。把它放在翻译任务入口就能天然控制并发度。importasyncioimportaiohttpimportaiofilesfrompathlibimportPath API_URLhttps://api.example.com/translateMAX_CONCURRENT5# 根据 API 限流调整semaphoreasyncio.Semaphore(MAX_CONCURRENT)asyncdeftranslate_one(session:aiohttp.ClientSession,file_path:str)-dict:单个文件翻译受 Semaphore 保护asyncwithsemaphore:# 同时最多 MAX_CONCURRENT 个协程进入asyncwithaiofiles.open(file_path,rb)asf:dataawaitf.read()formaiohttp.FormData()form.add_field(file,data,filenamePath(file_path).name)form.add_field(target_lang,zh)try:asyncwithsession.post(API_URL,dataform,timeout30)asresp:resp.raise_for_status()resultawaitresp.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_URLhttps://api.example.com/translateMAX_CONCURRENT5RETRY2semaphoreasyncio.Semaphore(MAX_CONCURRENT)asyncdeftranslate_with_retry(session,file_path:str,retries:intRETRY)-dict:带重试的单个文件翻译forattemptinrange(retries1):resultawaittranslate_one(session,file_path)ifresult[status]successorattemptretries:returnresultawaitasyncio.sleep(2**attempt)# 指数退避returnresultasyncdeftranslate_one(session:aiohttp.ClientSession,file_path:str)-dict:asyncwithsemaphore:try:asyncwithaiofiles.open(file_path,rb)asf:dataawaitf.read()formaiohttp.FormData()form.add_field(file,data,filenamePath(file_path).name)form.add_field(target_lang,zh)asyncwithsession.post(API_URL,dataform,timeout30)asresp:ifresp.status429: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_dirPath(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]resultsawaitasyncio.gather(*tasks)successsum(1forrinresultsifr[status]success)failedlen(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_workersN)不也能限流吗确实可以但两者有本质区别维度ThreadPoolExecutorasyncio.Semaphore并发模型多线程适合 CPU/IO 混合任务单线程协程适合高 IO 任务资源占用每个线程有独立栈空间数量多时有内存压力协程轻量可创建成千上万个适用场景PDF 解析等 CPU 密集型操作网络请求等 IO 密集型操作灵活性固定线程数调整不够细粒度可动态调整、可嵌套使用对于 PDF 翻译这种上传文件 → 等待 API 响应 → 下载结果的 IO 密集型任务asyncio Semaphore通常是更好的选择。五、生产环境进阶动态限流根据 API 返回的 429 频率自动调整 Semaphore 大小。队列化把任务先放入 Redis 队列消费端用 Semaphore 控制并发。超时与熔断连续失败超过阈值时暂停任务避免雪崩。连接池复用aiohttp.ClientSession不要每次请求都新建连接。六、总结批量 PDF 翻译不是并发越高越好。合理的并发控制能显著提升成功率和稳定性。asyncio.Semaphore是一个轻量、易用的限流工具配合重试和日志可以支撑大多数生产场景。如果你的翻译服务正在被 429 或内存耗尽困扰不妨先把并发度降下来把成功率提上去。标签Python、并发编程、asyncio、PDF翻译、API限流