如何解决aiogram+Flask Telegram机器人的业务逻辑阻塞问题？

你的问题核心在于：

• Flask默认是单线程同步服务器，同步阻塞操作（requests HTTP请求、googletrans同步翻译、频繁的数据库同步提交）会占用唯一的工作线程，导致后续请求排队等待
• 业务逻辑函数top_100_films_imdb全是同步操作，单次执行耗时30秒，完全阻塞了请求处理流程

1. 用异步IO替换所有同步阻塞操作 #

替换requests为aiohttp #

所有HTTP请求改用异步客户端，避免阻塞事件循环：

# 先安装依赖：pip install aiohttp
import aiohttp

async def top_100_films_imdb(bot, chat_id):
    url = "https://uk.wikipedia.org/wiki/Список_250_найрейтинговіших_фільмів_IMDb"
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            html = await response.text()
    soup = BeautifulSoup(html, 'lxml')

    films = soup.find('table', class_='sortable').find('tbody').findAll('tr', limit=101)
    # 后续循环内的requests.get也要改成aiohttp异步调用

替换同步翻译为异步实现 #

googletrans同步版本会阻塞，改用异步翻译方案，示例用aiohttp直接调用Google翻译API：

async def translate_text(session, text, dest='en'):
    url = "https://translate.googleapis.com/translate_a/single"
    params = {
        'client': 'gtx',
        'sl': 'auto',
        'tl': dest,
        'dt': 't',
        'q': text
    }
    try:
        async with session.get(url, params=params) as response:
            result = await response.json()
            return result[0][0][0], result[2]  # 返回翻译结果、源语言
    except Exception as e:
        print(f"翻译出错: {e}")
        return text, 'en'

# 业务函数内调用示例
async with aiohttp.ClientSession() as session:
    translated_text, src_lang = await translate_text(session, film_name, dest='en')

2. 把耗时业务逻辑放到异步任务队列 #

不要在Webhook请求处理函数里直接await耗时业务，用asyncio.create_task把任务丢到后台执行，让Webhook快速返回响应：
修改get_messages函数：

import asyncio

@app.route('/' + bot_token, methods=['GET', 'POST'])
async def get_messages():
    print("Flag")
    print(request.json)
    message_type = list(request.json.keys())[1]

    if message_type == 'message':
        # 用create_task替代await，后台执行任务
        asyncio.create_task(message(bot))
    elif message_type == 'callback_query':
        asyncio.create_task(callback_query(bot, payment_token, stripe_token))
    # pre_checkout_query这类需要即时响应的操作，仍需await
    elif message_type == 'pre_checkout_query':
        chat_id = request.json["pre_checkout_query"]["from"]["id"]
        await add_purchase(bot, chat_id)
        pre_checkout_query_id = request.json["pre_checkout_query"]["id"]
        await bot.answer_pre_checkout_query(pre_checkout_query_id, True)
    # 其他分支同理处理

    db.session.commit()
    return {"get messages is ok": True}

3. 优化数据库操作 #

减少频繁的db.session.commit()，把循环内的提交放到循环结束后批量执行，降低数据库IO阻塞：

async def top_100_films_imdb(bot, chat_id):
    # ... 爬取、解析逻辑
    for film in films:
        # ... 单条电影处理逻辑
        # 移除循环内的db.session.commit()
    
    # 所有电影处理完成后统一提交
    db.session.commit()

4. 用支持异步的Flask服务器 #

Flask默认开发服务器不支持异步，部署时使用ASGI服务器，比如gunicorn搭配uvicorn worker：

# 安装依赖：pip install gunicorn uvicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

多worker进程可进一步提升并发处理能力。

5. 其他优化点 #

• 不要在循环内重复创建Translator实例，提前在函数外初始化一次
• 对爬取的页面做缓存（比如用Redis），避免重复请求相同URL，有效期设为1天即可

import aiohttp
from bs4 import BeautifulSoup
import asyncio
import unidecode

async def translate_text(session, text, dest='en'):
    url = "https://translate.googleapis.com/translate_a/single"
    params = {
        'client': 'gtx',
        'sl': 'auto',
        'tl': dest,
        'dt': 't',
        'q': text
    }
    try:
        async with session.get(url, params=params) as response:
            result = await response.json()
            return result[0][0][0], result[2]
    except Exception as e:
        print(f"翻译出错: {e}")
        return text, 'en'

async def top_100_films_imdb(bot, chat_id):
    url = "https://uk.wikipedia.org/wiki/Список_250_найрейтинговіших_фільмів_IMDb"
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            html = await response.text()
    soup = BeautifulSoup(html, 'lxml')

    films = soup.find('table', class_='sortable').find('tbody').findAll('tr', limit=101)
    async with aiohttp.ClientSession() as session:
        for idx, film in enumerate(films):
            if idx == 0:
                continue

            film_place = int(film.findAll('td')[0].text)
            try:
                film_name = film.find('i').text.replace('\n', '').replace('·', '-').replace('—', '-')
                translated_name, src_lang = await translate_text(session, film_name, dest='en')
            except AttributeError:
                film_name = film.findAll('td')[2].text.replace('\n', '').replace('·', '-').replace('—', '-')
                translated_name, src_lang = await translate_text(session, film_name, dest='en')
            
            uk_film_link = "https://uk.wikipedia.org" + film.findAll('td')[1].find('a').get('href')
            film_year = int(film.findAll('td')[3].text)
            film_rate = float(film.findAll('td')[4].text.replace('\n', '').replace(',', '.'))

            if src_lang != 'en':
                if film_name.split('/')[0] == film_name:
                    async with session.get(uk_film_link) as response:
                        html = await response.text()
                    soup = BeautifulSoup(html, 'lxml')
                    en_film_link = soup.find('li', class_='interwiki-en').find('a').get('href')
                    async with session.get(en_film_link) as response:
                        html = await response.text()
                    soup = BeautifulSoup(html, 'lxml')
                    film_name = soup.find('th', class_='infobox-above').text.replace('·', '-').replace('—', '-')
                else:
                    translated_part, part_lang = await translate_text(session, film_name.split('/')[0], dest='en')
                    if part_lang == 'en':
                        film_name = film_name.split('/')[0][:-1]
                    else:
                        film_name = film_name.split('/')[1][1:]

            film_name = unidecode.unidecode(film_name)

            # 数据库查询与更新
            film = Films.query.filter_by(name=film_name, directed_at=film_year).first()
            film_availability = film is not None

            top_film = TopImdbFilms.query.filter_by(name=film_name, directed_at=film_year).first()
            if top_film is None:
                same_place_film = TopImdbFilms.query.filter_by(place=film_place).first()
                if same_place_film:
                    db.session.delete(same_place_film)
                top_imdb_film = TopImdbFilms(film_place, film_name, film_year, film_rate, film_availability)
                db.session.add(top_imdb_film)
            else:
                top_film.place = film_place

            print(film_place, film_name, film_year, film_rate, film_availability)
        
        # 统一提交数据库修改
        db.session.commit()
    
    # 任务完成后通知用户
    await bot.send_message(chat_id, "IMDb Top100电影数据更新完成！")

内容的提问来源于stack exchange，提问作者Tomas Angelo