如何在单线程VPS上为Aiogram Telegram Bot实现并行后台任务

核心问题根源 #

你当前的任务阻塞是因为time.sleep()是同步阻塞函数，在asyncio单线程事件循环中，它会占用整个线程，导致其他任务无法获得执行机会，最终所有任务串行执行而非并行。

优化步骤 #

1. 替换同步休眠为异步休眠

直接把time.sleep()换成asyncio.sleep()，这是asyncio原生的非阻塞休眠方法，会主动把控制权交还给事件循环，让其他任务可以并行运行。

示例改造：

# 原错误代码
import time
async def loop1():
    while True:
        # 检查Google Sheets逻辑
        time.sleep(60)

# 改造后代码
import asyncio
async def loop1():
    while True:
        # 检查Google Sheets逻辑
        await asyncio.sleep(60)

2. 确保Google Sheets操作异步化

如果你的Google Sheets操作使用的是同步库（比如gspread），即使休眠改了，请求过程还是会阻塞事件循环。解决方式有两种：

• 使用异步Google Sheets库，比如gspread_asyncio；
• 用asyncio.to_thread()把同步操作包装到线程池执行，避免阻塞事件循环。

示例：用asyncio.to_thread()包装同步操作

import gspread
async def check_sheet():
    # 定义同步的Sheet检查逻辑
    def sync_check():
        gc = gspread.service_account()
        sheet = gc.open("目标表格").sheet1
        return sheet.get_all_records()
    
    # 将同步操作放到线程池执行，不阻塞事件循环
    sheet_data = await asyncio.to_thread(sync_check)
    # 后续数据处理逻辑

3. 任务异常防护

给每个循环任务添加异常捕获，避免单个任务崩溃导致整个机器人进程挂掉，同时增加重试延迟，防止频繁报错。

示例：

async def loop1():
    while True:
        try:
            await check_sheet()
            await asyncio.sleep(60)
        except Exception as e:
            print(f"Loop1执行出错: {str(e)}")
            await asyncio.sleep(10)  # 出错后延迟10秒重试

4. 启动任务的正确姿势

在Aiogram的on_startup回调中创建异步任务，确保机器人启动时自动加载后台任务：

from aiogram import Bot, Dispatcher

async def on_startup(dp: Dispatcher):
    asyncio.create_task(loop1())
    asyncio.create_task(loop2())
    asyncio.create_task(loop3())

if __name__ == "__main__":
    bot = Bot(token="你的机器人TOKEN")
    dp = Dispatcher(bot)
    dp.startup.register(on_startup)
    dp.run_polling()

可选：复杂调度用APScheduler

如果任务需要更灵活的调度规则（比如每天固定时段执行、间隔执行），可以使用APScheduler的asyncio版本，简化任务管理：

from apscheduler.schedulers.asyncio import AsyncIOScheduler

async def check_sheet_task():
    # Google Sheets检查逻辑

async def on_startup(dp: Dispatcher):
    scheduler = AsyncIOScheduler(timezone="Asia/Shanghai")
    # 每60秒执行一次任务
    scheduler.add_job(check_sheet_task, "interval", seconds=60)
    # 每天上午9点执行一次任务
    scheduler.add_job(check_sheet_task, "cron", hour=9)
    scheduler.start()

内容的提问来源于stack exchange，提问作者lisa.smith