如何在Python aiogram3中将阻塞函数放入单独线程？

针对同步计算函数阻塞机器人事件循环的问题，可通过asyncio.run_in_executor()将阻塞函数放到单独线程执行，避免影响机器人处理其他请求。具体实现如下：

修改后的完整代码 #

import asyncio
from time import sleep
import logging
from concurrent.futures import ThreadPoolExecutor  # 导入线程池

from aiogram import Bot, Dispatcher, Router
from aiogram.filters import Command
from aiogram.types import Message


token = "<token>"

bot = Bot(token=token)
dispatcher = Dispatcher()
router = Router()
dispatcher.include_router(router)

# 初始化线程池（可选，不指定则用默认线程池）
executor = ThreadPoolExecutor(max_workers=4)


def block_function() -> str:
    sleep(2)
    return "Complete!"


@router.message(Command("start"))
async def start(message: Message) -> None:
    username = message.from_user.username
    await message.answer(f"Hello, {username}!")
    await message.answer("Start...")
    # 用run_in_executor将阻塞函数放到线程执行
    result = await asyncio.run_in_executor(executor, block_function)
    await message.answer(result)


@router.message(Command("escape"))
async def stop_polling(message: Message) -> None:
    await message.answer("Bye!")
    await dispatcher.stop_polling()
    executor.shutdown()  # 关闭线程池（程序结束时清理）


async def main() -> None:
    logging.basicConfig(level=logging.INFO)
    await dispatcher.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

关键说明 #

• asyncio.run_in_executor()会把传入的同步函数提交到线程池执行，主事件循环可继续处理其他用户请求，不会被阻塞。
• 显式初始化ThreadPoolExecutor可控制线程数量（比如根据计算任务的并发需求调整max_workers），若不指定第一个参数（传None），会使用asyncio默认的线程池。
• 程序结束时需调用executor.shutdown()清理线程池资源，避免内存泄漏。

内容的提问来源于stack exchange，提问作者name 0x0000000F