You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

Python3.6 Flask中通过HTTP请求触发非阻塞异步函数的实现疑问

解决Flask路由中后台异步执行任务的问题

首先,你遇到的核心问题是:Flask默认是同步WSGI框架,直接调用async def定义的协程不会自动执行,而且你需要路由先返回响应,再让任务在后台运行。你的思路方向是对的,但在Python 3.6.3的环境下,需要适配正确的异步/后台任务处理方式,下面给你两种可行的方案:

方案一:用线程实现简单后台任务(最推荐,适合你的场景)

因为你的任务本质是调用外部CLI脚本(通过subprocess),属于IO密集型操作,用线程来后台执行是最简单直接的方式,不需要引入asyncio的复杂度,Python 3.6完全支持。

修改你的代码如下:

import threading
from flask import Flask
import subprocess

app = Flask(__name__)

def run_background_job(params):
    # 建议用列表形式构造命令,避免shell=True带来的安全风险
    command = ["python", "your_target_script.py"] + params.split()
    proc = subprocess.Popen(command)
    
    # 这里执行你的环境变量修改、文件IO等操作
    # ...
    
    # 等待子进程完成
    proc.wait()
    
    # 任务完成后通知外部API
    notify_external_api_job_complete()

def notify_external_api_job_complete():
    # 实现你的外部API调用逻辑
    print("Job completed, notifying external API...")

@app.route('/execute_job')
def execute_job():
    params = "your_script_args_here"
    # 启动后台线程,daemon=True表示线程随主进程退出而结束(可选)
    threading.Thread(
        target=run_background_job,
        args=(params,),
        daemon=True
    ).start()
    
    # 立刻返回响应,线程在后台运行
    return 'Launched async job according to params, it is now running.'

为什么这个方案适合你?

  • 完全兼容Python 3.6.3,不需要额外依赖
  • 逻辑简单直观,不需要理解复杂的异步协程机制
  • subprocess.Popen本身会创建独立子进程,线程只是等待它完成,不会阻塞Flask的主线程

方案二:用asyncio实现异步后台任务(如果坚持用异步)

如果你想基于asyncio来实现,需要注意Python 3.6的asyncio没有asyncio.run()(这个是3.7新增的),而且Flask的同步路由需要手动把异步任务放到单独的线程里运行,避免阻塞响应返回。

修改后的代码示例:

import asyncio
import threading
from flask import Flask
import subprocess

app = Flask(__name__)

async def run_async_job(params):
    # 用asyncio的异步子进程API替代subprocess.Popen,更符合异步风格
    command = ["python", "your_target_script.py"] + params.split()
    proc = await asyncio.create_subprocess_exec(*command)
    
    # 执行你的环境变量修改、文件IO等操作
    # ...
    
    # 等待子进程完成(异步等待,不会阻塞事件循环)
    await proc.wait()
    
    # 异步通知外部API(建议用aiohttp这类异步HTTP库)
    await notify_external_api_job_complete()

async def notify_external_api_job_complete():
    # 示例:模拟异步API调用
    print("Async job completed, notifying external API...")

def run_async_task(coro):
    # 适配Python 3.6的事件循环运行逻辑
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(coro)
    loop.close()

@app.route('/execute_job')
def execute_job():
    params = "your_script_args_here"
    # 把异步任务放到单独线程里运行,让Flask主线程立刻返回响应
    threading.Thread(
        target=run_async_task,
        args=(run_async_job(params),),
        daemon=True
    ).start()
    
    return 'Launched async job according to params, it is now running.'

关键说明

  • async def定义协程本身没有错,但在同步Flask中必须手动启动事件循环,并且放到独立线程中,否则会阻塞路由的响应返回
  • Python 3.6的asyncio.create_subprocess_exec已经支持异步子进程操作,比直接用subprocess.Popen更适合异步场景

额外注意事项

  1. 避免shell=True:构造subprocess命令时,尽量用列表形式,不要用字符串加shell=True,防止命令注入安全风险
  2. 任务持久化:如果你的后台任务需要在Flask服务重启后不丢失,或者需要更复杂的任务调度(比如重试、优先级),可以考虑用Celery这类专业的任务队列,但对于简单场景,线程/asyncio足够
  3. 资源清理:如果后台任务涉及文件句柄、数据库连接等资源,要确保在任务完成后正确关闭,避免资源泄漏

内容的提问来源于stack exchange,提问作者Spencer MacBeth

火山引擎 最新活动