Flask-SocketIO实现Python多线程与Angular实时通信异常排查

你的问题核心在于主线程被thread.join()阻塞，导致Flask-SocketIO的事件循环无法实时处理子线程的emit请求，所有消息被缓存到所有线程执行完毕后才一次性发送。

问题根源 #

在beginSimulation函数中，你启动所有子线程后立即调用了thread.join()，这会阻塞当前的SocketIO事件处理线程。而Flask-SocketIO的消息发送依赖事件循环调度，主线程被卡住时，所有子线程的emit操作都无法被及时处理，只能等到join完成后批量发送。

解决方案 #

移除事件处理函数中的join阻塞，改用SocketIO提供的start_background_task来后台等待线程完成，同时确保子线程能实时调用emit。

修改后的代码示例

main.py

from flask import Flask
from flask_cors import CORS
from flask_socketio import SocketIO
import threading

from Services.runSimulation import SimThread

app = Flask(__name__)
app.config['SECRET_KEY'] = 'vnkdjnfjknfl1232#'
socketio = SocketIO(app, cors_allowed_origins="*")
CORS(app)


@socketio.on('begin')
def beginSimulation(message, methods=['GET', 'POST']):
    print(message)
    socketio.emit('message_log', {'message': 'Simulation started.'})
    threads = []
    
    # 筛选有效节点并初始化屏障
    valid_nodes = [node for node in Example.vect 
                   if node not in Example.exits and node.id not in Example.blocked]
    barrier = threading.Barrier(len(valid_nodes))

    for node in valid_nodes:
        # 传入正确的socketio实例，设置为守护线程
        thread = SimThread(node, barrier, socketio)
        thread.daemon = True
        thread.start()
        threads.append(thread)
    
    # 启动后台任务等待所有线程完成，避免阻塞事件循环
    def wait_for_threads():
        for thread in threads:
            thread.join()
        socketio.emit('message_log', {'message': 'Simulation complete.'})
    
    socketio.start_background_task(wait_for_threads)


if __name__ == '__main__':
    socketio.run(app, debug=True)

simulation.py

import threading

class SimThread(threading.Thread):
    def __init__(self, node, barrier, socket):
        super().__init__()
        self.node = node
        self.capacity = node.capacity
        self.lock = threading.Lock()
        self.barrier = barrier
        self.socket = socket
        self.population = node.population  # 修正为实例变量，确保线程安全

    def run(self):
        self.barrier.wait()
        while self.population > 0:
            # 模拟逻辑（省略）
            ...
            
            # 使用SocketIO提供的sleep，避免阻塞事件循环
            self.socket.sleep(self.node.cost)
            
            # 仅在需要保护共享数据时使用锁，emit本身是线程安全的
            with self.lock:
                # 假设best_route和move_amount已在模拟逻辑中计算
                self.socket.emit('message_log', {
                    'from': self.node.id,
                    'to': best_route.id,
                    'pop': str(move_amount)
                })
                self.population -= move_amount  # 更新当前节点人口

关键修改点说明 #

1. 移除事件处理函数中的join：让beginSimulation函数立即返回，SocketIO事件循环可以继续处理消息发送。
2. 使用start_background_task：后台等待所有线程完成，完成后发送模拟结束消息，不阻塞主线程。
3. 设置守护线程：确保线程不会阻止程序正常退出。
4. 修正变量作用域：将population改为实例变量，避免线程间共享全局变量导致的问题。
5. 使用SocketIO的sleep：替代time.sleep，确保不会阻塞SocketIO的事件循环。

Angular端验证 #

确保组件中正确订阅消息：

import { Component, OnInit } from '@angular/core';
import { RequestsService } from './requests.service';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html'
})
export class AppComponent implements OnInit {
  constructor(private requestsService: RequestsService) {}

  ngOnInit(): void {
    this.requestsService.receive().subscribe(msg => {
      console.log('Received message:', msg);
      // 实时更新UI逻辑
    });
  }

  startSimulation(): void {
    this.requestsService.testpost();
  }
}

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