如何实现Telegram Bot与数据解析器同时独立运行？

问题核心 #

当前代码中，在Bot启动函数start_bot里直接await gis_pars()会导致解析器的同步阻塞代码（如Selenium操作、sleep()）卡住整个asyncio事件循环，Bot无法处理用户消息；反之如果先跑解析器，Bot也无法启动。本质是没有让两个任务在事件循环中并行独立运行，且解析器的同步代码没有做异步兼容处理。

解决方案 #

核心思路：

1. 将解析器中的同步阻塞代码（Selenium、sleep()）转移到后台线程执行，避免阻塞asyncio事件循环
2. 用asyncio.create_task()将解析器作为后台异步任务启动，与Bot的消息处理并行运行

步骤1：改造解析器代码 #

把Selenium的同步操作封装到单独的同步函数中，用asyncio.to_thread()（Python3.9+支持）在后台线程执行；将sleep()替换为异步的await asyncio.sleep()，确保解析器整体是异步友好的：

import asyncio
from selenium.common import StaleElementReferenceException, NoSuchElementException
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from time import sleep
from database.Database import DataBase
from gis.pathes import *

chrome_options = Options()
chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument("--disable-extensions")
chrome_options.add_argument("--start-maximized")

# 把所有同步阻塞的解析逻辑放到这个函数里
def run_selenium_parsing(db):
    driver = webdriver.Chrome(options=chrome_options)
    try:
        # 此处放原来的解析器工作代码：打开页面、提取数据、写入数据库等
        driver.get("目标采集URL")
        # 示例操作：提取元素、处理数据
        # title = driver.find_element(By.XPATH, title_path).text
        # db.insert_data(...)
        sleep(2)  # 原同步sleep保留在这里，因为函数在后台线程执行
    except (StaleElementReferenceException, NoSuchElementException) as e:
        print(f"解析出错: {e}")
    finally:
        driver.quit()

async def gis_pars():
    db = DataBase()
    # 持续采集的循环（可根据需求调整退出条件）
    while True:
        # 在后台线程运行同步解析代码
        await asyncio.to_thread(run_selenium_parsing, db)
        # 异步等待下一次采集（比如每小时一次）
        await asyncio.sleep(3600)

步骤2：修改Bot启动逻辑 #

在start_bot中用asyncio.create_task()启动解析器，而非直接await，让解析器在后台独立运行：

from aiogram import Dispatcher, Bot, F
import asyncio
import os
from dotenv import load_dotenv
from aiogram.client.bot import DefaultBotProperties
from aiogram.enums.parse_mode import ParseMode

# 导入路由和解析器
from gis.gis import gis_pars
from handlers.profile.profile import profile_router, process_pre_checkout_query, success_payment
from handlers.search_org.search_org import search_router
from handlers.add_organization.add_organization import create_route
from handlers.history_search.history_search import history_router
from handlers.mailing.mailing import mailing_route
from handlers.start.start import start_router
from handlers.view_user.view_user import view_router

load_dotenv()
token = os.getenv('TOKEN_ID')
bot = Bot(token=token, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher()

async def start_bot(bot: Bot):
    await bot.send_message(chat_id=*****, text='Bot已启动，解析器开始采集数据')
    # 创建后台异步任务启动解析器，不阻塞Bot的消息处理
    asyncio.create_task(gis_pars())

# 注册路由和回调
dp.startup.register(start_bot)
dp.include_router(start_router)
dp.include_router(profile_router)
dp.include_router(search_router)
dp.include_router(create_route)
dp.include_router(history_router)
dp.include_router(mailing_route)
dp.include_router(view_router)

dp.pre_checkout_query.register(process_pre_checkout_query)
dp.message.register(success_payment, F.successful_payment)

async def start():
    try:
        await dp.start_polling(bot, skip_updates=True)
    finally:
        await bot.session.close()

if __name__ == '__main__':
    asyncio.run(start())

关键注意事项 #

• 如果数据库操作是同步的，也要放到run_selenium_parsing这类后台线程函数中，避免阻塞事件循环
• 可根据需求调整解析器的循环间隔（asyncio.sleep()的参数），或添加停止条件（比如检测特定信号）
• 建议给解析器的任务添加异常捕获，避免解析器崩溃导致整个程序退出

内容的提问来源于stack exchange，提问作者Ирина Брунёва