Flask-SocketIO后台无限循环任务阻塞队列读取问题求助

核心问题 #

代码阻塞的根源有两点：

1. SocketIO事件处理函数长期阻塞：grab_raw_data作为WebSocket事件处理函数，启动后台任务后进入while True循环，占用了整个eventlet协程，导致后台任务和其他SocketIO事件无法得到调度。
2. 非协程友好的工具使用：标准库queue.Queue的阻塞操作不兼容协程模型，会阻塞eventlet的事件循环；后台任务中使用的time.sleep虽被monkey patch替换，但直接用eventlet原生sleep更可靠。

另外，后台任务代码存在笔误：data.put应改为data_queue.put，否则会抛出未定义错误。

解决方案步骤 #

1. 替换为协程友好的队列

将标准库queue.Queue替换为eventlet.queue.Queue，它的阻塞操作会挂起协程而非占用线程，适配eventlet异步模型。

2. 拆分队列监听逻辑

把队列监听的while True循环从事件处理函数中移出，放到独立的后台任务中，让事件处理函数尽快返回，释放协程资源。

3. 使用eventlet原生sleep

将后台任务中的time.sleep替换为eventlet.sleep，确保协程能正确让出CPU时间。

修改后的代码示例 #

main.py

import eventlet
from eventlet.queue import Queue  # 替换标准库Queue
from threading import Lock

from get_raw_program import get_raw_data
from flask import Flask
from flask_socketio import SocketIO

eventlet.patcher.monkey_patch(select=True, socket=True)

app = Flask(__name__, static_folder=static_folder, static_url_path='')
app.config['SECRET_KEY'] = 'secret_key'

socketio = SocketIO(app, binary=True, async_mode="eventlet", logger=True, engineio_logger=True)
thread = None
listener_thread = None
thread_lock = Lock()
data_queue = Queue()

def queue_listener():
    """独立的队列监听后台任务"""
    while True:
        # 阻塞等待队列数据，协程会挂起而非占用线程
        data = data_queue.get(block=True)
        socketio.emit('raw_data', {'msg': data})
        # 让出CPU时间，让其他协程运行
        socketio.sleep(0.0001)

@socketio.on('WebSocket_On')
def grab_raw_data(test):
    global thread, listener_thread

    with thread_lock:
        if thread is None:
            socketio.emit('raw_data', {'msg': 'Thread is None:'})
            socketio.emit('raw_data', {'msg': 'Starting Thread... '})
            # 启动数据抓取后台任务
            thread = socketio.start_background_task(target=get_raw_data, args=(data_queue, test['mode']))
            # 启动队列监听后台任务
            listener_thread = socketio.start_background_task(target=queue_listener)

get_raw_program.py

import eventlet

def get_raw_data(data_queue, test):
    count = 0
    while True:
        # 修正笔误：data_queue而非data
        data_queue.put(b'\xe5\xce\x04\x00\xfe\xd2\x04\x00')
        # 使用eventlet原生sleep，协程友好
        eventlet.sleep(0.001)
        count += 1

额外说明 #

• eventlet的协程模型依赖非阻塞操作和主动让出CPU时间，所有长期运行的逻辑都必须放在start_background_task创建的后台任务中。
• 若需要停止后台任务，可设置全局标志变量，在循环中检查标志来退出，例如：

  # 在main.py中添加全局标志
  stop_flag = False
  
  # 修改queue_listener和get_raw_data中的循环条件
  while not stop_flag:
      # ... 原有逻辑
  

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