顾乔芝士网

持续更新的前后端开发技术栈

Flask应用中处理异步事件(后台线程+事件循环)的方法

在我们的实际应用当中,我们需要定义三个东西

一个多线程的信号旗,这里的信号旗指的是多线程编程中一个非常基础且重要的同步工具

它只有两种状态:

Clear (清除/未设置): 信号灯是红色的,旗子是放下的。

Set (设置): 信号灯是绿色的,旗子是举起的。

当进程调用event.wait(),如果此时事件是 “Clear” 状态(红灯),那么这个线程就会停下来,进入等待状态,被阻塞。

一个 循环的对象

一个新线程

接下来在这里我们看到了,有一个target参数,这个就是调用我们的一个说明书

# 1. 创建一个线程同步工具

mcp_ready_event = threading.Event()


# 2. 创建一个新的 asyncio 事件循环对象

background_event_loop = asyncio.new_event_loop()

# 3. 创建一个新线程,目标是运行 start_event_loop 函数

loop_thread = threading.Thread(

target=start_event_loop,

args=(background_event_loop, mcp_ready_event),

daemon=True

name="name"

)

loop_thread.start() # 启动这个新线程


# 4. 主线程在这里暂停,等待后台线程通知

logging.info("主线程正在等待 MCP 初始化...")

is_ready = mcp_ready_event.wait(timeout=30.0) # 暂停30秒,等待加载


那么我们的说明书怎么写呢?这里举例mcp的连接初始化

初始化这里,无论成功还是失败,初始化流程走完后,调用 .set() 来释放正在 wait() 的主线程。

# ... existing code ...

async def initialize_mcp_connection(ready_event: threading.Event):

"""

在应用启动时启动MCP服务器并建立一个持久的会话。

完成后通过 event 通知主线程。

"""

# ...

try:

# 初始化代码


except Exception as e:

# ... 错误处理 ...

finally:

logging.info("MCP 初始化过程结束,通知主线程。")

ready_event.set()


初始化完毕之后,我们就进行接下来的开始线程

这里的参数 loop: asyncio.AbstractEventLoop 和 ready_event: threading.Event 是什么意思?

这是一种现代 Python 的语法,叫做 类型提示 (Type Hinting)。它本身不强制任何东西,但它极大地增强了代码的可读性和可维护性,并且能被很多开发工具利用来进行代码检查。

loop:
asyncio.AbstractEventLoop: 这行代码的意思是,我们期望 loop 这个参数是一个 asyncio.AbstractEventLoop 类型的对象。它告诉任何阅读或调用这个函数的人:“请给我一个 asyncio 的事件循环对象。”

ready_event: threading.Event: 同理,这表示我们期望 ready_event 这个参数是一个我们刚才讲的 threading.Event 类型的对象。

def start_event_loop(loop: asyncio.AbstractEventLoop, ready_event: threading.Event):

"""在一个线程中启动事件循环,并在启动时安排MCP初始化。"""

asyncio.set_event_loop(loop)

loop.create_task(initialize_mcp_connection(ready_event))

logging.info("后台事件循环已启动,MCP 初始化任务已安排。")

loop.run_forever()


需要注意这里start_event_loop 函数中的 loop,和主线程中的 background_event_loop,虽然名字不同,但它们指向的是内存中同一个事件循环对象。

最后只需要运行一个专门用于从一个普通线程向一个正在运行 asyncio 事件循环的线程提交任务。

asyncio.run_coroutine_threadsafe(coro, loop):

控制面板
您好,欢迎到访网站!
  查看权限
网站分类
最新留言