如何在Python Flask中每60秒执行玩家操作并更新所有用户页面？

方法1：Celery 实现定时任务（推荐生产环境） #

Celery是分布式任务队列，适合多进程/多机器场景，和Flask配合需要单独启动Celery worker，步骤如下：

1. 配置Celery和Flask
   先安装依赖：pip install celery redis（用Redis做消息 broker，也可替换为RabbitMQ）
   在Flask项目中创建Celery实例：

   from flask import Flask
   from celery import Celery
   
   app = Flask(__name__)
   app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
   app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'
   
   def make_celery(app):
       celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
       celery.conf.update(app.config)
       return celery
   
   celery = make_celery(app)
   
   # 定义每60秒执行的同步任务
   @celery.task
   def sync_player_actions():
       # 业务逻辑示例：
       # 1. 从SQL库读取所有玩家操作，未选择则用默认值
       # 2. 批量执行操作计算游戏状态
       # 3. 将新状态写回数据库
       print("正在同步玩家操作...")
   

2. 配置周期性任务
   创建celery_config.py文件：

   from datetime import timedelta
   
   CELERY_BEAT_SCHEDULE = {
       'sync-every-60s': {
           'task': '你的Flask模块名.sync_player_actions',
           'schedule': timedelta(seconds=60),
       },
   }
   

   在Flask app中加载配置：app.config.from_pyfile('celery_config.py')

3. 启动服务
   需要同时启动三个独立进程：

   • Flask应用：flask run
   • Celery worker：celery -A 你的Flask模块名.celery worker --loglevel=info
   • Celery beat（负责触发定时任务）：celery -A 你的Flask模块名.celery beat --loglevel=info

方法2：APScheduler 轻量定时任务（适合小型项目） #

APScheduler是轻量级定时框架，可直接嵌入Flask，无需额外启动进程（生产环境建议用独立进程部署）

1. 安装依赖：pip install apscheduler

2. 集成到Flask

   from flask import Flask
   from apscheduler.schedulers.background import BackgroundScheduler
   
   app = Flask(__name__)
   
   def sync_player_actions():
       # 同上述业务逻辑
       print("正在同步玩家操作...")
   
   # 初始化调度器
   scheduler = BackgroundScheduler()
   scheduler.add_job(sync_player_actions, 'interval', seconds=60)
   scheduler.start()
   
   # Flask退出时关闭调度器
   @app.route('/')
   def index():
       return "游戏服务器运行中"
   
   if __name__ == '__main__':
       try:
           app.run(debug=True)
       except (KeyboardInterrupt, SystemExit):
           scheduler.shutdown()
   

   注意：用Gunicorn等生产服务器时，需确保调度器只启动一次，可通过--workers=1或配置Redis分布式锁避免多进程重复执行任务。

方法3：Asyncio 循环（适合异步Flask场景） #

如果使用Flask 2.0+异步支持或FastAPI，可直接用asyncio实现定时任务：

from flask import Flask
import asyncio

app = Flask(__name__)

async def sync_player_actions():
    while True:
        # 同上述业务逻辑
        print("正在同步玩家操作...")
        await asyncio.sleep(60)

# Flask首次请求前启动异步任务
@app.before_first_request
def start_async_task():
    loop = asyncio.get_event_loop()
    loop.create_task(sync_player_actions())

@app.route('/')
async def index():
    return "游戏服务器运行中"

if __name__ == '__main__':
    app.run(debug=True)

注意：异步Flask生产部署需用Uvicorn等支持异步的服务器。

关键注意事项 #

• 数据库并发控制：同步操作时需用事务或行级锁，避免多进程同时修改数据导致状态不一致。
• 默认操作处理：查询玩家操作时，可通过SQL的COALESCE(action, '默认操作')直接处理未选择的情况。
• 任务幂等性：确保任务重复执行时不会重复处理同一周期的操作，可通过在数据库记录周期标识实现。

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