如何在Python Flask中处理并发写操作导致的数据冲突-Python教程-PHP中文网

并发写丢数据是因为全局变量、字典和文件操作非线程安全，多请求同时修改导致覆盖；应使用数据库行级锁（如SELECT ... FOR UPDATE）、Redis原子命令或正确实现乐观锁并处理冲突。

如何在python flask中处理并发写操作导致的数据冲突

Flask 应用里直接用全局变量或文件存数据时，为什么并发写会丢数据

因为 Python 的 global 变量、内置字典、甚至普通文件写入（如 open(..., 'w')）都不是线程安全的。多个请求同时执行 data['count'] += 1 或 f.write()，底层可能只读到旧值、覆盖彼此修改，最终结果小于预期——这不是 Flask 的锅，是共享状态没加锁。

常见现象包括：计数器跳变、JSON 文件写乱码、用户提交后数据“消失”或重复写入。尤其在开发模式下用 flask run（默认多线程）或部署到 Gunicorn（多 worker）时更容易暴露。

别用 dict / list 等内存结构存跨请求状态；临时缓存可用 threading.local()，但不解决跨进程冲突
文件写入必须加锁：threading.Lock() 只管同进程内，多 worker 需 filelock 或外部协调服务
若用 SQLite，注意它默认只支持一个写连接，高并发下会抛 OperationalError: database is locked

用 SQLAlchemy + 数据库行级锁避免更新丢失

最稳妥的方式是把状态交给数据库，并用 SELECT ... FOR UPDATE 显式加锁。Flask-SQLAlchemy 支持 with_for_update()，能保证“读-改-写”原子性。

例如用户积分扣减场景：

from sqlalchemy import text
<p>@app.route('/deduct', methods=['POST'])
def deduct_points():
user_id = request.json['user_id']
amount = request.json['amount']</p><pre class="brush:php;toolbar:false;"># 加锁查余额（阻塞直到锁释放）
user = User.query.with_for_update().filter_by(id=user_id).first()
if not user or user.points <p></p>

PostgreSQL/MySQL 支持 with_for_update()；SQLite 不支持，会静默忽略（务必确认 DB 类型）
锁范围默认是查到的行，但若 WHERE 条件无索引，可能升级为表锁，拖慢整体性能
锁持有时间越短越好——别在 with_for_update() 块里调外部 API 或做耗时计算

用 Redis 实现带重试的原子计数器

当需要高频、轻量的并发写（如限流、投票、PV 统计），Redis 的单线程模型和原子命令是更优解。Python 用 redis-py 调用 INCR、HINCRBY 等命令天然避开了应用层竞争。

Python Hardener

通过完整的安全审查、代码整洁重构、健壮的错误处理、正确的日志记录以及完整的文档字符串覆盖，强化 Python 脚本。

下载

复杂逻辑（如“余额充足才扣减”）需用 Lua 脚本保证原子性：

# Lua 脚本：检查并扣减，失败返回 nil
lua_script = """
if redis.call('HGET', KEYS[1], 'balance') >= tonumber(ARGV[1]) then
  redis.call('HINCRBY', KEYS[1], 'balance', -ARGV[1])
  return 1
else
  return nil
end
"""
<h1>Python 调用</h1><p>result = redis.eval(lua_script, 1, f"user:{user_id}", amount)
if result is None:
abort(400, "balance insufficient")
</p>