You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在同一Python文件中独立同时运行Socket服务器与Flask应用

如何在同一Python文件中独立同时运行Socket服务器与Flask应用

嗨,我完全懂你的困扰——你想让Socket服务器和Flask应用在同一个Python文件里同时跑起来,之前试了线程但没成功:Socket服务器一启动,Flask就没动静,关掉Socket后Flask才开始工作。这是因为两个服务都是阻塞式的,得把它们放到独立的线程里,还要处理好数据共享的问题,我来给你一步步解决:

核心问题分析

你的原代码里,server()函数是个死循环(while True),如果直接在主线程跑它,程序会一直卡在这个循环里,根本轮不到执行Flask的app.run()。而Flask的run()方法默认也是阻塞的,所以必须把其中一个(或两个)放到后台线程中运行。

另外,你需要把Socket收到的数据展示在Flask里,所以还要准备一个线程安全的共享数据容器,避免多线程操作数据时出现混乱。

完整可运行代码示例

下面是整合后的代码,我标注了关键细节:

import socket
import threading
from flask import Flask

# 线程安全的共享数据:存储Socket收到的消息
received_data = []
# 锁对象,保护共享数据的读写,避免多线程冲突
data_lock = threading.Lock()

def socket_server():
    host = socket.gethostname()
    port = 8080 

    server_socket = socket.socket()  
    server_socket.bind((host, port)) 
    server_socket.listen(2)
    conn, address = server_socket.accept()  
    print("Connection from: " + str(address))

    while True:
        data = conn.recv(1024).decode()
        # 处理关闭连接的情况
        if str(data) == "Connection closed" or not data:
            break
        print("Received data: " + str(data))
        # 加锁更新共享数据
        with data_lock:
            received_data.append(data)
    
    conn.close() 
    print("Socket connection closed")

# 初始化Flask应用
app = Flask(__name__)

@app.route("/data")
def data_from_server():
    # 读取共享数据时也要加锁,保证数据一致性
    with data_lock:
        # 将列表转为易读的字符串返回,无数据时给出提示
        return "\n".join(received_data) if received_data else "No data received yet"

if __name__ == '__main__':
    # 启动Socket服务器线程,设置daemon=True:主线程退出时,该线程自动终止
    socket_thread = threading.Thread(target=socket_server, daemon=True)
    socket_thread.start()
    
    # 启动Flask应用
    # 注意:debug模式下要加use_reloader=False,否则Flask会重启进程,导致启动多个Socket服务器
    app.run(host='0.0.0.0', port=5000, debug=True, use_reloader=False)

关键细节说明

  • 线程守护模式(daemon=True):设置后,当Flask主线程退出时,Socket后台线程会自动跟着终止,不会留在后台占用资源。
  • 线程安全锁:用threading.Lock()确保同一时间只有一个线程读写共享数据,避免出现数据丢失或混乱的情况。
  • Flask的use_reloader参数:如果开启debug模式,Flask默认会重新加载代码并启动新进程,这会导致再次执行socket_thread.start(),创建多个Socket服务器。加上use_reloader=False可以避免这个问题。

测试方法

  1. 运行这个Python文件,你会看到Socket服务器启动,同时Flask应用也会在http://localhost:5000运行。
  2. 用客户端工具(比如telnet或者自己写一个简单的Socket客户端)连接到你的Socket服务器(地址是socket.gethostname()返回的主机名,端口8080),发送一些消息。
  3. 访问http://localhost:5000/data,就能看到Socket收到的所有消息啦!

备注:内容来源于stack exchange,提问作者Filip craqow

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.14 11:23:07