如何在Python-Flask框架中运行永久后台进程并实现玩家连接通知

一、将后台游戏进程整合进Flask应用 #

不用单独启动独立进程，直接用Python的threading模块在Flask启动时开启一个后台线程运行游戏主循环即可。这种方式能让游戏进程和Flask服务共享同一个游戏实例，无需额外处理进程间通信，实现起来更简便。

修改你的现有代码如下：

import sys
import json
import threading
import time
from flask import Flask, request, render_template

app = Flask(__name__)

from mf import gameplay

game = gameplay.Game()

# 后台游戏循环函数
def game_loop():
    while True:
        # 这里编写游戏事件推进逻辑，比如更新全局状态、处理AI行为、触发时间驱动事件等
        game.update_game_state()
        # 控制游戏更新频率，比如每秒执行一次
        time.sleep(1)

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

@app.route('/api', methods=['POST'])
def api():
    data_json = request.data.decode()
    data = json.loads(data_json)
    return game(data_json)

if __name__ == '__main__':
    # 启动后台游戏线程，daemon=True让线程随Flask主进程退出而终止
    game_thread = threading.Thread(target=game_loop, daemon=True)
    game_thread.start()
    # 启动Flask服务
    app.run('0.0.0.0', 5000, threaded=True)

注意：需要在你的gameplay.Game类中实现update_game_state()方法，用来处理无玩家连接时的游戏事件推进逻辑。

二、玩家（重新）连接时通知后台进程 #

有两种简便方式实现这个需求，根据你的场景选择即可：

方式1：直接在路由中触发游戏回调 #

在玩家访问的路由（首页或API接口）中，直接调用游戏实例的对应方法，触发玩家连接后的特定操作：

@app.route('/')
def index():
    # 通知游戏有玩家连接
    game.handle_player_connect()
    return render_template('mainmap.html')

@app.route('/api', methods=['POST'])
def api():
    data_json = request.data.decode()
    data = json.loads(data_json)
    # 针对玩家重连的请求，触发专属逻辑
    if data.get('action') == 'reconnect':
        game.handle_player_reconnect(data.get('player_id'))
    return game(data_json)

然后在gameplay.Game类中实现handle_player_connect()和handle_player_reconnect()方法，比如同步最新游戏状态给玩家、触发欢迎事件、恢复玩家之前的进度等。

方式2：用线程事件触发后台响应 #

如果需要后台游戏线程立即暂停常规逻辑，优先处理玩家连接事件，可以用threading.Event实现：

import threading

# 初始化游戏和连接事件
game = gameplay.Game()
player_connect_trigger = threading.Event()

def game_loop():
    while True:
        # 检查是否有玩家连接触发信号
        if player_connect_trigger.is_set():
            # 执行玩家连接后的专属操作
            game.process_player_connect()
            # 重置信号，等待下一次触发
            player_connect_trigger.clear()
        # 游戏常规状态更新
        game.update_game_state()
        time.sleep(1)

@app.route('/')
def index():
    # 触发玩家连接信号
    player_connect_trigger.set()
    return render_template('mainmap.html')

这种方式适合需要后台线程立即响应玩家连接的场景，比如实时同步玩家数据、触发即时互动事件。

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