如何在Flask应用中同时运行OBS WebSocket连接与静态网页端点？

我需要实现两个功能：

• 连接OBS WebSocket服务器并监听事件
• 为静态网页创建Flask端点

单独运行connect_to_obs函数时能正常获取OBS的所有事件，但无法让它和Flask应用同时运行，求解决办法。

我的Flask应用代码 #

from flask import Flask, request, session, jsonify, make_response, Response, send_from_directory, render_template, redirect, url_for
from flask_cors import CORS
from apscheduler.schedulers.background import BackgroundScheduler
import os
from flask_cors import CORS, cross_origin
from functions import createFolder, fileHandler, update_obs_settings, get_profile_image, get_obs_settings, searchCategories, validateAccessToken, checkCredentials, createTags, storeTokenResponse, init_db, getAccessTokenFromDb, store_user_data
from OBSConnection import connect_to_obs
import time
import json
import requests
import threading
import asyncio

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'

scheduler = BackgroundScheduler()

init_db()

app.config['UPLOAD_FOLDER'] = uploadFolder


t = threading.Thread(target=connect_to_obs)
t.start()

@app.route('/')
def index():
    return render_template('index.html')

# main driver function
if __name__ == '__main__':
    t = threading.Thread(target=connect_to_obs)
    t.start()
    app.run(debug=True)

我的connect_to_obs函数代码 #

def connect_to_obs():
    obs_url, obs_port, obs_password = get_obs_settings()
    if obs_url != None or obs_port != None:
        parameters = simpleobsws.IdentificationParameters() # Create an IdentificationParameters object
        parameters.eventSubscriptions = None # Subscribe to the General and Scenes categories
        print(f'{obs_url}:{obs_port} password: {obs_password}')
        ws = simpleobsws.WebSocketClient(url=f'ws://{obs_url}:{obs_port}', password=obs_password) # Every possible argument has been passed, but none are required. See lib code for defaults.

        async def onReplayBufferSave(eventType, eventData):
            if eventType == 'ReplayBufferSaved':
                print(eventData['savedReplayPath'])

        async def onRecordStateChanged(eventType, eventData):
            if eventType == 'RecordStateChanged':
                print(eventData['outputPath'])

        async def init():
            global obsConnection
            while True:
                try:
                    await ws.connect()
                    await ws.wait_until_identified()
                    ws.register_event_callback(onReplayBufferSave)
                    ws.register_event_callback(onRecordStateChanged)
                    print('Connected to OBS Websocket')
                    obsConnection = True
                    break
                except:
                    print("Connection failed, attempting to reconnect in 5 seconds...")
                    await asyncio.sleep(5)
        loop = asyncio.get_event_loop()
        loop.run_until_complete(init())
        loop.run_forever()
    else:
        print("Please set up OBS settings in the settings page.")
        exit()

解决方案 #

问题核心：子线程直接复用主线程的asyncio事件循环，导致与Flask的运行逻辑冲突；同时存在重复启动线程、debug模式双进程重复加载的问题。

修改后的Flask主文件 #

from flask import Flask, render_template
from flask_cors import CORS
from apscheduler.schedulers.background import BackgroundScheduler
import os
from functions import init_db, get_obs_settings
from OBSConnection import connect_to_obs
import threading
import asyncio

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'

scheduler = BackgroundScheduler()

init_db()

# 定义上传文件夹路径（根据实际情况调整）
uploadFolder = './uploads'
app.config['UPLOAD_FOLDER'] = uploadFolder

@app.route('/')
def index():
    return render_template('index.html')

if __name__ == '__main__':
    # 避免debug模式下双进程重复启动线程
    if not app.debug or os.environ.get('WERKZEUG_RUN_MAIN') == 'true':
        # 设置守护线程，Flask退出时自动终止
        obs_thread = threading.Thread(target=connect_to_obs, daemon=True)
        obs_thread.start()
    app.run(debug=True)

修改后的connect_to_obs函数 #

import simpleobsws

def connect_to_obs():
    obs_url, obs_port, obs_password = get_obs_settings()
    # 修正判断逻辑，确保URL和端口都有效
    if obs_url and obs_port:
        parameters = simpleobsws.IdentificationParameters()
        # 明确订阅所有事件（替代原None设置，避免订阅不全）
        parameters.eventSubscriptions = simpleobsws.EventSubscription.All
        print(f'{obs_url}:{obs_port} password: {obs_password}')
        ws = simpleobsws.WebSocketClient(url=f'ws://{obs_url}:{obs_port}', password=obs_password)

        async def onReplayBufferSave(eventType, eventData):
            if eventType == 'ReplayBufferSaved':
                print(eventData['savedReplayPath'])

        async def onRecordStateChanged(eventType, eventData):
            if eventType == 'RecordStateChanged':
                print(eventData['outputPath'])

        async def init():
            global obsConnection
            obsConnection = False
            while not obsConnection:
                try:
                    await ws.connect()
                    await ws.wait_until_identified()
                    ws.register_event_callback(onReplayBufferSave)
                    ws.register_event_callback(onRecordStateChanged)
                    print('Connected to OBS Websocket')
                    obsConnection = True
                    # 连接成功后保持事件循环运行，处理OBS推送的事件
                    await ws.loop.run_forever()
                except Exception as e:
                    print(f"连接失败: {str(e)}, 5秒后重试...")
                    obsConnection = False
                    await asyncio.sleep(5)
                finally:
                    # 断开连接时清理资源
                    if ws.is_connected():
                        await ws.disconnect()

        # 在子线程中创建独立的asyncio事件循环，避免与主线程冲突
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(init())
    else:
        print("请在设置页面配置OBS连接参数。")

关键修改点 #

1. 独立事件循环：子线程中用asyncio.new_event_loop()创建专属循环，避免和Flask主线程的循环冲突
2. 守护线程：设置daemon=True，Flask退出时子线程自动终止，无需手动处理
3. debug模式适配：通过WERKZEUG_RUN_MAIN环境变量判断是否为主进程，避免双进程重复启动线程
4. 逻辑优化：修正OBS连接的判断逻辑，明确事件订阅范围，添加异常捕获的具体信息

内容的提问来源于stack exchange，提问作者Dad The Gamer