如何编写可定时执行任务且不阻塞其他功能的Telegram机器人

问题分析 #

• 方案一的while True循环会完全阻塞主线程，导致机器人无法响应任何Telegram消息请求。
• 方案二的核心缺陷是await dp.start_polling()为阻塞式调用，执行后后续的调度器初始化代码永远不会被触发，定时任务根本无法启动。

正确实现方案 #

以下是基于aiogram + AsyncIOScheduler的异步正确实现，确保实时消息处理与定时任务并行运行：

import asyncio
import json
import logging
from apscheduler.schedulers.asyncio import AsyncIOScheduler

from aiogram import Bot, Dispatcher, types
from aiogram.utils.exceptions import BotBlocked

logging.basicConfig(level=logging.INFO)

# 配置参数
BOT_TOKEN = '你的机器人Token'
CHANNELS_FILE = "channels.json"
MESSAGE_TEXT = "这是定时推送的测试消息！"

# 初始化Bot与Dispatcher
bot = Bot(token=BOT_TOKEN)
dp = Dispatcher(bot)

# 加载已保存的频道ID集合
try:
    with open(CHANNELS_FILE, 'r', encoding='utf-8') as f:
        channel_ids = set(json.load(f))
except FileNotFoundError:
    channel_ids = set()

# 处理/start命令
@dp.message_handler(commands=['start'])
async def start(message: types.Message):
    await message.reply(
        "你好！我是支持定时推送的机器人。\n把我添加到频道，我会记录频道ID并定时推送消息。"
    )

# 监听频道消息，自动记录新添加的频道ID
@dp.channel_post_handler()
async def channel_post(message: types.Message):
    channel_id = message.chat.id
    if channel_id not in channel_ids:
        channel_ids.add(channel_id)
        with open(CHANNELS_FILE, 'w', encoding='utf-8') as f:
            json.dump(list(channel_ids), f, ensure_ascii=False)
        print(f"已记录新频道ID: {channel_id}")

# 定时任务核心逻辑：向所有记录的频道发送消息
async def send_scheduled_message():
    if not channel_ids:
        logging.info("无已记录频道，跳过本次推送")
        return
    
    # 遍历集合副本，避免遍历过程中集合修改引发异常
    for channel_id in channel_ids.copy():
        try:
            await bot.send_message(channel_id, MESSAGE_TEXT)
            logging.info(f"消息已推送至频道: {channel_id}")
        except BotBlocked:
            logging.warning(f"机器人被频道 {channel_id} 拉黑，已移除该频道")
            channel_ids.remove(channel_id)
            with open(CHANNELS_FILE, 'w', encoding='utf-8') as f:
                json.dump(list(channel_ids), f, ensure_ascii=False)
        except Exception as e:
            logging.error(f"推送消息至频道 {channel_id} 失败: {str(e)}")

async def main():
    # 初始化异步调度器，设置时区（根据需求修改）
    scheduler = AsyncIOScheduler(timezone="Asia/Shanghai")
    # 示例：每隔10秒推送一次，可改为每日固定时间：scheduler.add_job(..., 'cron', hour=21, minute=0)
    scheduler.add_job(send_scheduled_message, 'interval', seconds=10)
    scheduler.start()
    logging.info("定时任务调度器已启动")

    # 将消息轮询转为异步任务，避免阻塞调度器运行
    polling_task = asyncio.create_task(dp.start_polling())
    await polling_task

if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logging.info("机器人已停止运行")

关键改进点 #

• 使用asyncio.create_task包装dp.start_polling()，让消息轮询在后台运行，不再阻塞调度器初始化。
• 添加异常处理逻辑，处理机器人被频道拉黑的情况，避免定时任务崩溃。
• 遍历频道集合的副本，防止遍历过程中集合被修改引发错误。
• 设置时区参数，确保定时任务时间准确。
• 优化文件读写的编码配置，避免中文乱码问题。

注意事项 #

• 确保安装依赖：pip install aiogram apscheduler
• 替换BOT_TOKEN为你的机器人实际Token
• 可根据需求修改定时任务规则，例如每日21点执行：

  scheduler.add_job(send_scheduled_message, 'cron', hour=21, minute=0, timezone="Asia/Shanghai")
  

内容的提问来源于stack exchange，提问作者Artem Shvalev