如何在Flask中运行异步函数？Telegram通知报错求助

报错原因分析 #

1. asyncio.create_task 报错：Flask同步视图运行在同步上下文中，没有活跃的asyncio事件循环，而create_task必须在已启动的事件循环环境下调用，直接使用会触发上下文错误。
2. asyncio.run 报错：你的异步Telegram Bot已经绑定了一个事件循环，asyncio.run会创建全新的独立事件循环，导致Bot的send_message操作跨循环执行，触发超时上下文相关的RuntimeError。

可行解决方案 #

核心思路：在Flask返回响应后，通过后台线程启动独立的asyncio事件循环来执行异步通知任务，既不阻塞响应返回，也避免事件循环冲突。

步骤1：添加异步任务执行的线程封装函数 #

在代码中导入threading和asyncio，新增用于在线程中运行异步协程的工具函数：

import threading
import asyncio

def run_async_coro(coro):
    # 创建新的事件循环并运行协程
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(coro)
    finally:
        loop.close()

步骤2：修改Flask视图中的异步任务调用 #

将原有的asyncio.create_task替换为线程启动：

@app.route('/', methods=['POST', 'GET'])
def webhook():
    data = request.form.to_dict()
    if 'event' in data and data['event'] == 'ONTASKCOMMENTADD':
        task_id = data['data[FIELDS_AFTER][TASK_ID]']
        com_id = data['data[FIELDS_AFTER][ID]']
        url = url_crm + "task.commentitem.get"
        headers = CaseInsensitiveDict()
        headers["Accept"] = "application/json"
        headers["Content-Type"] = "application/json"
        data = {"TASKID":task_id,"ITEMID":com_id}
        
        maker_id = get_bitrix24_tgid(get_bitrix24_task(task_id)['RESPONSIBLE_ID'])
        resp = requests.post(url, headers=headers, json=data).json()['result']
        files = []
        if 'ATTACHED_OBJECTS' in resp:
            for key in resp['ATTACHED_OBJECTS']:
                files.append(get_link(resp['ATTACHED_OBJECTS'][key]['FILE_ID']))
        
        # 用后台线程执行异步通知任务，daemon=True确保线程随主进程退出
        threading.Thread(
            target=run_async_coro,
            args=(notify_user(resp['AUTHOR_NAME'], maker_id, task_id, resp['POST_MESSAGE'], files),),
            daemon=True
        ).start()
        
    return jsonify({'status': 'ok'})

步骤3：优化异步函数中的同步调用（可选） #

如果get_bitrix24_task是同步阻塞函数，建议将其包装到线程池中执行，避免阻塞异步事件循环：

async def notify_user(author, user_id, task_id, text, links):
    print(author, user_id, task_id, text, links)
    if len(links) == 0:
        # 用run_in_executor将同步函数放到线程池执行
        loop = asyncio.get_event_loop()
        task = await loop.run_in_executor(None, get_bitrix24_task, task_id)
        print(task)
        if task != 'Error':
            keyboard = InlineKeyboardMarkup()
            if int(task['REAL_STATUS']) <= 2:
                start_data = 'start_' + task_id 
                start_btn = InlineKeyboardButton(text='starts', callback_data=start_data)
                keyboard.add(start_btn)
            elif task['REAL_STATUS'] == '3':
                stop_data = 'stop_' + task_id 
                stop_btn = InlineKeyboardButton(text='stopp', callback_data=stop_data)
                keyboard.add(stop_btn)
            com_data = 'com_' + task_id 
            com_btn = InlineKeyboardButton(text='comm', callback_data=com_data)
            keyboard.add(com_btn)
            temp_txt = '111'
            await bot.send_message(int(user_id), temp_txt)

关键说明 #

• daemon=True：标记线程为守护线程，当Flask主进程退出时，后台线程会自动终止，避免残留僵尸线程。
• 独立事件循环：每个后台线程创建自己的事件循环，与Telegram Bot的主循环完全隔离，解决跨循环调用的冲突问题。

内容的提问来源于stack exchange，提问作者Эльдар Шигапов