如何在多线程调用的外部模块中通过Flask-SocketIO发送数据？

方案1：将socketio对象作为参数传入外部函数 #

这是最直接且低耦合的实现方式，修改外部模块的函数，让它接收socketio实例作为参数：

修改后的process_module.py #

import time

def start_process(socketio):
    for i in range(50):
        message = "Hello, world!"
        socketio.emit('message', message)
        print(f"Event triggered! {message}")
        time.sleep(1)

修改后的main_script.py #

from flask import Flask
from flask_socketio import SocketIO
import threading
from process_module import start_process

app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="")

if __name__ == "__main__":
    # 启动线程时传入socketio实例
    threading.Thread(target=start_process, args=(socketio,)).start()
    socketio.run(app, port=1120, debug=False)

方案2：在外部模块中延迟导入主模块的socketio #

如果不想修改函数参数，可以在外部模块的函数内部导入主模块的socketio，避免循环导入问题：

修改后的process_module.py #

import time

def start_process():
    # 函数内部延迟导入，避免循环导入
    from main_script import socketio
    for i in range(50):
        message = "Hello, world!"
        socketio.emit('message', message)
        print(f"Event triggered! {message}")
        time.sleep(1)

main_script.py保持不变 #

from flask import Flask
from flask_socketio import SocketIO
import threading
from process_module import start_process

app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="")

if __name__ == "__main__":
    threading.Thread(target=start_process).start()
    socketio.run(app, port=1120, debug=False)

注意：这种方式耦合度较高，主模块名称固定，后续修改主模块文件名需要同步调整导入路径。

方案3：使用回调函数解耦 #

如果希望外部模块完全不依赖Flask-SocketIO，可以定义一个发送消息的回调函数，传给start_process：

修改后的process_module.py #

import time

def start_process(send_message):
    for i in range(50):
        message = "Hello, world!"
        send_message(message)
        print(f"Event triggered! {message}")
        time.sleep(1)

修改后的main_script.py #

from flask import Flask
from flask_socketio import SocketIO
import threading
from process_module import start_process

app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="")

def send_socket_message(msg):
    socketio.emit('message', msg)

if __name__ == "__main__":
    threading.Thread(target=start_process, args=(send_socket_message,)).start()
    socketio.run(app, port=1120, debug=False)

三种方案中，方案1和方案3更推荐，它们降低了模块间的耦合度，代码更易维护。

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