Flask会话在Google App Engine上无法持久化的国际象棋游戏问题

问题根源 #

你用全局字典instances存储游戏会话，但Google App Engine是多实例分布式环境，每个实例拥有独立内存空间。当负载均衡把用户请求路由到不同实例时，新实例的instances字典中没有之前存储的会话数据，就会出现会话丢失。本地单实例运行正常，是因为所有请求共享同一内存空间。

无额外成本的替代方案 #

1. 利用Flask内置会话存储（适合小体量数据） #

Flask的session默认会把加密后的数据存在客户端Cookie中，你可以将SomeClass的关键数据序列化后存入session，无需自行维护instances字典。

修改后的代码示例：

from flask import Flask, session
import uuid
import pickle
from flask_cors import CORS

app = Flask(__name__, template_folder=".")
# 生产环境务必使用随机生成的安全密钥，可从GAE环境变量读取
app.secret_key = "your_secure_production_secret_key"
# 开启credentials支持，允许客户端携带Cookie
CORS(app, supports_credentials=True)

@app.route("/new-session")
def new_session():
    game_id = str(uuid.uuid4())
    game_instance = SomeClass(game_id)
    # 将游戏实例序列化后存入session，需确保SomeClass可被pickle序列化
    session['game_data'] = pickle.dumps(game_instance)
    return {"status": "session_created"}

# 走棋接口示例
@app.route("/make-move", methods=["POST"])
def make_move():
    game_data = session.get('game_data')
    if not game_data:
        return {"error": "会话已丢失"}, 400
    
    game_instance = pickle.loads(game_data)
    # 在这里处理走棋逻辑，比如game_instance.make_move(...)
    
    # 更新后重新存入session
    session['game_data'] = pickle.dumps(game_instance)
    return {"move_result": "success"}

注意：

• Cookie有4KB左右的大小限制，所以SomeClass的实例数据不能过大
• 若pickle序列化有问题，可自行将实例转换为字典格式存储
• 生产环境的secret_key绝对不能硬编码，建议通过GAE的环境变量配置

2. 使用Google Cloud Datastore（免费额度覆盖小应用需求） #

Cloud Datastore提供免费额度（每日5万次读操作、2万次写操作），完全满足简易国际象棋游戏的需求。可以用session_id作为主键，将会话数据存储在Datastore中。

代码示例：

from flask import Flask, session
import uuid
import pickle
from datetime import datetime
from google.cloud import datastore
from flask_cors import CORS

app = Flask(__name__, template_folder=".")
app.secret_key = "your_secure_production_secret_key"
CORS(app, supports_credentials=True)

# 初始化Datastore客户端，GAE部署时会自动关联项目
datastore_client = datastore.Client()

@app.route("/new-session")
def new_session():
    session_id = str(uuid.uuid4())
    game_instance = SomeClass(session_id)
    
    # 创建Datastore实体，以session_id为键
    entity = datastore.Entity(key=datastore_client.key("GameSession", session_id))
    entity.update({
        "game_data": pickle.dumps(game_instance),
        "created_at": datetime.now()
    })
    datastore_client.put(entity)
    
    # 将session_id存入Flask session，让客户端携带
    session['session_id'] = session_id
    return {"session_id": session_id}

@app.route("/make-move", methods=["POST"])
def make_move():
    session_id = session.get('session_id')
    if not session_id:
        return {"error": "会话已丢失"}, 400
    
    # 从Datastore获取会话数据
    key = datastore_client.key("GameSession", session_id)
    entity = datastore_client.get(key)
    if not entity:
        return {"error": "会话不存在"}, 400
    
    game_instance = pickle.loads(entity["game_data"])
    # 处理走棋逻辑
    
    # 更新后重新存入Datastore
    entity["game_data"] = pickle.dumps(game_instance)
    datastore_client.put(entity)
    return {"move_result": "success"}

注意：

• 部署前需在Google Cloud控制台启用Cloud Datastore
• 可添加过期逻辑，定期清理闲置会话，避免占用存储空间

可选方案：使用Redis（性能需求较高时） #

如果游戏数据量大或对读写性能要求高，可选用Google Cloud Memorystore（托管Redis服务），它提供免费试用额度，小应用的运维成本极低。

代码示例：

from flask import Flask, session
import uuid
import pickle
import os
import redis
from flask_cors import CORS

app = Flask(__name__, template_folder=".")
app.secret_key = "your_secure_production_secret_key"
CORS(app, supports_credentials=True)

# 从GAE环境变量读取Redis连接信息
redis_client = redis.Redis(
    host=os.environ.get("REDIS_HOST"),
    port=int(os.environ.get("REDIS_PORT", 6379)),
    password=os.environ.get("REDIS_PASSWORD")
)

@app.route("/new-session")
def new_session():
    session_id = str(uuid.uuid4())
    game_instance = SomeClass(session_id)
    # 设置会话1小时过期，可根据需求调整
    redis_client.setex(session_id, 3600, pickle.dumps(game_instance))
    
    session['session_id'] = session_id
    return {"session_id": session_id}

@app.route("/make-move", methods=["POST"])
def make_move():
    session_id = session.get('session_id')
    if not session_id:
        return {"error": "会话已丢失"}, 400
    
    game_data = redis_client.get(session_id)
    if not game_data:
        return {"error": "会话已过期"}, 400
    
    game_instance = pickle.loads(game_data)
    # 处理走棋逻辑
    
    # 更新会话并重置过期时间
    redis_client.setex(session_id, 3600, pickle.dumps(game_instance))
    return {"move_result": "success"}

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