Flask应用与后台线程间Python字典同步异常排查求助

你遇到的锁失效和数据不一致问题，大概率是以下几个原因导致的，对应解决方案如下：

1. Flask多进程部署导致全局变量不共享 #

如果你的Flask应用是通过多进程方式部署的（比如Gunicorn多worker、uWSGI多进程，甚至Flask内置服务器开启processes参数），那么每个进程都会拥有独立的folder_access_times字典和data_lock锁实例——也就是说，请求线程修改的是当前进程的字典，而后台线程读取的是另一个进程的字典，自然看不到更新，锁也完全起不到跨进程同步的作用。

解决方案： #

• 使用进程间共享存储：
  • 用multiprocessing.Manager创建跨进程共享的字典和锁：

    from multiprocessing import Manager
    manager = Manager()
    folder_access_times = manager.dict()
    data_lock = manager.Lock()
    

    所有进程都使用这个共享的字典和锁，就能保证数据一致性。
  • 或者改用外部存储（如Redis）：将用户访问时间存在Redis中，并设置30秒的过期时间，这样无需自己编写清理线程，Redis会自动删除过期的键值对，同时天然支持多进程/多实例共享。

2. 线程内存可见性与锁的使用细节 #

即使是单线程部署，Python的线程也可能因为局部缓存导致看不到其他线程的修改（虽然GIL保证执行顺序，但内存可见性仍需同步机制保障）。另外，你的代码中存在几个潜在的锁使用问题：

潜在问题与修复： #

• 时间计算错误：(curtime - val).seconds只会返回时间差的秒部分，若时间差超过1分钟（如61秒），会被截断为1秒，导致误判。应改用total_seconds()：

  diff = (curtime - val).total_seconds()
  if diff > 30:
      shutil.rmtree(...)
  

• 遍历字典时的并发修改风险：即使在锁内遍历folder_access_times.items()，如果遍历过程中其他线程修改了字典（比如删除键），可能引发异常。建议先拷贝键的列表再遍历：

  with data_lock:
      curtime = datetime.datetime.now()
      # 拷贝键列表，避免遍历中字典结构变化
      for key in list(folder_access_times.keys()):
          val = folder_access_times[key]
          diff = (curtime - val).total_seconds()
          if diff > 30:
              del folder_access_times[key]
              shutil.rmtree(f"static/files/userdata_{key}/")
  

• 确保所有字典操作都在锁内：检查代码中是否有其他地方（比如处理文件上传、删除时）读取或修改folder_access_times，这些操作必须同样包裹在with data_lock:块中。

3. Flask内置服务器的线程模型验证 #

如果你用的是Flask默认的开发服务器，默认是单线程模式——此时后台线程和请求线程属于同一进程，但如果没有正确启动后台线程，或者线程没有持续运行，也可能导致问题。确保后台线程是在应用启动后持续运行的，比如：

import threading
import time

def cleanup_task():
    while True:
        with data_lock:
            curtime = datetime.datetime.now()
            for key in list(folder_access_times.keys()):
                val = folder_access_times[key]
                if (curtime - val).total_seconds() > 30:
                    del folder_access_times[key]
                    shutil.rmtree(f"static/files/userdata_{key}/")
        time.sleep(10)  # 每隔10秒检查一次，无需高频轮询

# 在应用启动时启动后台线程
if __name__ == "__main__":
    threading.Thread(target=cleanup_task, daemon=True).start()
    app.run(threaded=True)  # 开启多线程处理请求

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