如何为Python Telegram Bot实现盲文图片翻译的并发处理？

我用python-telegram-bot开发了一款Bot，功能是接收盲文图片，翻译后返回英文结果。核心翻译方法定义如下：

async def start_translation(update: Update, context: CallbackContext):
    # 翻译逻辑（耗时较长）
    return translated_text

当前存在的问题是：start_translation执行耗时过长，单用户请求处理时，其他用户发送的图片请求必须等待前一个翻译完成才能开始处理。之前尝试用threading模块，但它不支持直接运行async函数。翻译流程的启动代码如下：

def create_handle_photo(oVoice, oCorrect, oInference, current_dir):
    async def handle_photo(update: Update, context: CallbackContext):
        image_path = await get_image(update, context, current_dir) # 下载图片
        oProcessImage = ProcessImage(image_path)
        text = await start_translation(update, context, oProcessImage, oInference, oCorrect)
        if text != '': 
            await generate_voice(update, context, current_dir, oVoice, text)
        await clean_up(current_dir, image_path) # 删除下载的文件
    return handle_photo

def main():
   app = ApplicationBuilder().token(TOKEN).build()
   handle_photo_handler = create_handle_photo(oVoice, oCorrect, oInference, current_dir)
   app.add_handler(MessageHandler(filters.PHOTO, handle_photo_handler)) # 处理图片消息
   app.run_polling()

if __name__ == '__main__':
    main()

Bot输入输出示例：

解决方案：利用asyncio的run_in_executor实现并行处理 #

python-telegram-bot基于asyncio事件循环，默认单线程执行，耗时的同步操作会阻塞整个循环导致请求串行。我们可以把翻译核心逻辑抽成同步函数，放到线程池/进程池里执行，让事件循环能同时处理多个用户请求。

步骤1：抽离同步翻译逻辑

把start_translation里的耗时翻译逻辑单独写成纯同步函数（去掉async/await）：

def sync_translation_logic(oProcessImage, oInference, oCorrect):
    # 原start_translation中的所有翻译逻辑（纯同步代码）
    # 比如图片识别、盲文转英文等耗时操作
    return translated_text

步骤2：改造start_translation函数

用asyncio.run_in_executor把同步逻辑放到线程池执行，避免阻塞事件循环：

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 全局或在main中创建线程池，根据并发需求设置最大线程数
translation_executor = ThreadPoolExecutor(max_workers=8)

async def start_translation(update: Update, context: CallbackContext, oProcessImage, oInference, oCorrect):
    loop = asyncio.get_event_loop()
    # 在线程池中执行同步翻译逻辑
    translated_text = await loop.run_in_executor(
        translation_executor,
        sync_translation_logic,
        oProcessImage, oInference, oCorrect
    )
    return translated_text

如果翻译是CPU密集型任务，建议改用ProcessPoolExecutor规避GIL限制：

from concurrent.futures import ProcessPoolExecutor

translation_executor = ProcessPoolExecutor(max_workers=4)

步骤3：修改启动代码传递执行器

调整create_handle_photo和main函数，把执行器传入处理逻辑：

def create_handle_photo(oVoice, oCorrect, oInference, current_dir, executor):
    async def handle_photo(update: Update, context: CallbackContext):
        image_path = await get_image(update, context, current_dir)
        oProcessImage = ProcessImage(image_path)
        text = await start_translation(update, context, oProcessImage, oInference, oCorrect, executor)
        if text != '': 
            await generate_voice(update, context, current_dir, oVoice, text)
        await clean_up(current_dir, image_path)
    return handle_photo

def main():
    # 初始化执行器
    translation_executor = ThreadPoolExecutor(max_workers=8)
    app = ApplicationBuilder().token(TOKEN).build()
    handle_photo_handler = create_handle_photo(oVoice, oCorrect, oInference, current_dir, translation_executor)
    app.add_handler(MessageHandler(filters.PHOTO, handle_photo_handler))
    app.run_polling()

if __name__ == '__main__':
    main()

注意事项 #

• 若使用ProcessPoolExecutor，确保sync_translation_logic中用到的对象（如oProcessImage）是可序列化的，否则会报错。
• 线程/进程池的max_workers值根据服务器CPU核心数和预期并发量调整，避免资源过载。

内容的提问来源于stack exchange，提问作者Chandler Bong