如何在Flet中实现浏览数的跨用户实时同步更新？

你的while True会直接阻塞Flet的主事件循环，导致页面彻底失去响应。要实现浏览数实时同步且不影响交互，得利用Flet的异步/定时任务特性，同时解决文件读写的并发冲突问题。

关键优化点 #

• 避免阻塞事件循环：用Flet的Timer组件定时拉取最新数据，而非死循环
• 解决并发冲突：给文件操作加锁，防止多用户同时修改导致数据错乱
• 复用UI组件：固定一个文本组件，直接更新内容而非重复添加

修改后的完整代码 #

import flet as ft
from filelock import FileLock  # 先安装依赖：pip install filelock

def main(page: ft.Page):
    # 创建全局可更新的浏览数文本组件
    views_display = ft.Text("Views: 0", size=20)
    page.add(views_display)

    def load_latest_count():
        """读取文件中的最新浏览数并更新UI"""
        with FileLock("views.lock"):
            try:
                with open("views", "r") as f:
                    count = int(f.read().strip() or 0)
                views_display.value = f"Views: {count}"
                page.update()
                return count
            except FileNotFoundError:
                # 文件不存在时初始化计数
                with open("views", "w") as f:
                    f.write("0")
                views_display.value = "Views: 0"
                page.update()
                return 0

    def increment_count():
        """递增浏览数并写入文件"""
        current_count = load_latest_count()
        with FileLock("views.lock"):
            with open("views", "w") as f:
                f.write(str(current_count + 1))
        load_latest_count()

    # 页面加载时先获取一次最新计数
    load_latest_count()
    # 用户访问时触发计数递增
    page.on_route_change = lambda e: increment_count()

    # 定时任务：每2秒自动同步最新计数
    def sync_count(_):
        load_latest_count()

    page.add(ft.Timer(2000, sync_count, repeat=True))

if __name__ == "__main__":
    ft.app(target=main)

代码说明 #

1. 文件锁：通过FileLock保证同一时间只有一个进程能读写计数文件，彻底避免多用户并发修改导致的数值错误
2. 定时同步：用Timer每2秒自动拉取文件中的最新计数，所有在线用户的页面都会实时更新
3. UI复用：固定一个Text组件，每次只更新它的value属性，避免页面组件冗余
4. 路由触发：用户访问页面时触发路由变化事件，执行计数递增逻辑

无额外依赖替代方案 #

如果不想安装filelock，可以用Python内置线程锁（仅适用于单进程部署场景）：

import flet as ft
import threading

file_lock = threading.Lock()

def main(page: ft.Page):
    views_display = ft.Text("Views: 0", size=20)
    page.add(views_display)

    def load_latest_count():
        with file_lock:
            try:
                with open("views", "r") as f:
                    count = int(f.read().strip() or 0)
                views_display.value = f"Views: {count}"
                page.update()
                return count
            except FileNotFoundError:
                with open("views", "w") as f:
                    f.write("0")
                views_display.value = "Views: 0"
                page.update()
                return 0

    def increment_count():
        current_count = load_latest_count()
        with file_lock:
            with open("views", "w") as f:
                f.write(str(current_count + 1))
        load_latest_count()

    load_latest_count()
    page.on_route_change = lambda e: increment_count()

    def sync_count(_):
        load_latest_count()

    page.add(ft.Timer(2000, sync_count, repeat=True))

if __name__ == "__main__":
    ft.app(target=main)

注意：如果是多进程部署（比如Gunicorn多worker），线程锁会失效，建议改用Redis、SQLite等数据库存储计数，更适合分布式场景。

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