Flet PubSubClient 完全指南：Python 跨会话实时消息发布与订阅

前端跨平台桌面应用移动开发【免费下载链接】fletBuild realtime web, mobile and desktop apps in Python only. No frontend experience required.项目地址https://gitcode.com/gh_mirrors/fl/flet点击查看免费下载在 Flet 中每个用户打开应用都会对应一个独立的会话Session不同会话之间默认互不可见。如果你需要构建聊天室、实时通知、协作编辑这类一个用户发消息、所有其他用户立刻看到的功能就需要一套跨会话的通信机制。Flet 提供了内置的PubSub发布/订阅组件而 PubSubClient 正是这套机制中面向应用开发者最常用的入口——它是绑定到当前页面会话的客户端门面所有发布、订阅、退订操作都围绕当前会话自动完成。读完本文你将掌握如何通过page.pubsub获取客户端、send_all/send_others/subscribe/subscribe_topic等全部方法的语义与适用场景、一个可直接运行的聊天应用完整示例以及底层PubSubHub的实现原理与会话清理机制。PubSubClient 是什么会话级的 PubSub 门面PubSubClient是 Flet 内置 PubSub 机制中绑定到单个会话的客户端对象。从源码的类注释可以直接看到它的定位pubsub_client.pySession-scoped facade overPubSubHub. This client binds all pub/sub operations to one session ID so callers can publish and subscribe without passing their session identity explicitly on each call.翻译过来就是它是PubSubHub的会话级门面Facade。所有 pub/sub 操作都被绑定到一个 session ID 上调用者在每次调用时无需手动传递会话身份。构造函数需要两个参数pubsub_client.pydef __init__(self, pubsub: PubSubHub, session_id: str):pubsub进程内共享的PubSubHub实例负责真正的消息路由session_id当前会话的唯一标识send_others发给别人、unsubscribe按会话退订等操作都依赖它来区分自己和别人。在常规的 Flet 服务端应用中你不需要手动构造PubSubClient。会话建立时Flet 会在 session.py 中自动完成装配self.__pubsub_client PubSubClient(conn.pubsubhub, self.__id)其中self.__id是通过random_string(16)生成的 16 位随机会话 IDsession.pyconn.pubsubhub则是连接层持有的进程级PubSubHub。而Page对象暴露了pubsub属性直接透传会话客户端page.pyproperty def pubsub(self) - PubSubClient: The PubSub client for the current page. return self.session.pubsub_client所以在应用代码里拿到当前页面客户端只需一行client page.pubsub九个核心方法全解析PubSubClient共提供 9 个方法可分为发布、订阅、退订三大类。下表汇总了全部方法与语义分类方法作用回调参数发布send_all(message)向所有会话广播全局消息—发布send_all_on_topic(topic, message)向指定主题的所有订阅者广播—发布send_others(message)向除本会话外的所有会话广播全局消息—发布send_others_on_topic(topic, message)向主题订阅者广播但排除本会话—订阅subscribe(handler)订阅全局广播消息(message)订阅subscribe_topic(topic, handler)订阅某个主题(topic, message)退订unsubscribe()移除本会话的全部全局订阅—退订unsubscribe_topic(topic)移除本会话对指定主题的订阅—退订unsubscribe_all()移除本会话所有全局 主题订阅—发布消息send_all 系列四个发送方法全部转发到共享的PubSubHub其中send_others系列由客户端自动补上本会话 ID 实现排除自己pubsub_client.pydef send_all(self, message: Any): self.__pubsub.send_all(message) def send_all_on_topic(self, topic: str, message: Any): self.__pubsub.send_all_on_topic(topic, message) def send_others(self, message: Any): self.__pubsub.send_others(self.__session_id, message) def send_others_on_topic(self, topic: str, message: Any): self.__pubsub.send_others_on_topic(self.__session_id, topic, message)message的类型是Any可以是字符串、字典、列表甚至是自定义对象。注意send_others系只排除发起者所在会话其他所有会话哪怕也有多个浏览器标签都会收到。订阅消息subscribe 与 subscribe_topicsubscribe(handler)注册当前会话的全局消息处理器处理器只接收一个位置参数messagepubsub_client.pydef subscribe(self, handler: Callable[[Any], Any]): The handler is invoked with one positional argument: message. self.__pubsub.subscribe(self.__session_id, handler)subscribe_topic(topic, handler)则注册主题订阅处理器接收两个位置参数(topic, message)pubsub_client.pydef subscribe_topic(self, topic: str, handler: Callable[[str, Any], Any]): The handler is invoked with two positional arguments: (topic, message). self.__pubsub.subscribe_topic(self.__session_id, topic, handler)从PubSubHub的实现看处理器既可以是普通同步函数也可以是async协程函数pubsub_hub.py。同步函数在事件循环的线程池执行器中运行异步函数则通过run_coroutine_threadsafe调度回事件循环——这意味着你可以在回调里安全地操作 UI 控件并调用page.update()。退订unsubscribe 系列三个退订方法的粒度不同pubsub_client.pyunsubscribe()只清掉本会话的全局广播订阅主题订阅不受影响unsubscribe_topic(topic)只清掉本会话对指定主题的订阅unsubscribe_all()一次性清掉本会话的全部全局与主题订阅。典型使用流程订阅 → 发布 → 退订官方 PubSub 教程 给出了一个典型的生命周期范式核心步骤恰好覆盖了客户端的三大类方法应用会话启动时调用subscribe()订阅广播或subscribe_topic()订阅主题某个事件发生时如发送按钮点击调用send_all()广播或send_all_on_topic()按主题发送某个事件发生时如离开按钮点击调用unsubscribe()或unsubscribe_topic()页面关闭时在page.on_close中调用unsubscribe_all()清理一切订阅。实战示例一个基于 PubSub 的完整聊天应用下面这个聊天应用来自官方教程但特意用到了send_others只发给别人自己那条消息由本地逻辑直接追加避免自己发自己收带来的重复渲染问题。它展示了订阅、发布、退订的完整闭环import flet as ft def main(page: ft.Page): page.title Flet Chat # 1. 订阅全局广播消息别人发来的消息追加到消息列表 def on_message(msg): messages.controls.append(ft.Text(msg)) page.update() page.pubsub.subscribe(on_message) # 2. 发送消息发给所有其他会话自己本地直接显示 def send_click(e): page.pubsub.send_others(f{user.value}: {message.value}) messages.controls.append(ft.Text(f{user.value}: {message.value})) message.value page.update() # 3. 离开聊天退订广播消息 def leave_click(e): page.pubsub.unsubscribe() messages.controls.append(ft.Text(You left the chat)) page.update() messages ft.Column() user ft.TextField(hint_textYour name, width150) message ft.TextField(hint_textYour message..., expandTrue) send ft.Button(Send, on_clicksend_click) leave ft.Button(Leave, on_clickleave_click) page.add( messages, ft.Row(controls[user, message, send, leave]), ) ft.run(main, viewft.AppView.WEB_BROWSER)两个浏览器窗口打开同一应用即为两个会话在其中一个发送消息另一个会实时收到——这正是PubSubClient最典型的落地场景效果见下图底层原理PubSubHub 的线程安全路由PubSubClient只是薄薄的一层门面真正的消息路由在PubSubHub中完成pubsub_hub.pyThread-safe in-memory pub/sub router scoped to a Flet server process.理解它有助于把握客户端行为的上限与边界。核心要点如下1. 三份索引结构pubsub_hub.pyself.__subscribers: dict[str, set[handler]] # session_id - 全局handler集合 self.__topic_subscribers: dict[str, dict[str, set]] # topic - session_id - handler集合 self.__subscriber_topics: dict[str, dict[str, set]] # session_id - topic - handler集合反向索引前两份用于高效路由发布第三份反向索引让按会话退订unsubscribe_all能在常数时间内找到该会话订阅过的全部主题。2. 线程安全整个 hub 用一把threading.Lock保护所有读写。在 Pyodide浏览器内运行环境下则替换为无操作锁NopeLock因为该环境不存在多线程竞争pubsub_hub.py。3. 同步与异步处理器分派pubsub_hub.pyif inspect.iscoroutinefunction(handler): asyncio.run_coroutine_threadsafe(handler(*args), self.__loop) else: if self.__executor: self.__loop.call_soon_threadsafe( self.__loop.run_in_executor, self.__executor, handler, *args ) else: handler(*args)异步 handlerrun_coroutine_threadsafe提交到事件循环执行同步 handler有 executor 时提交到线程池避免阻塞事件循环无 executor 时在当前线程内联调用未配置事件循环任何发送操作都会抛出RuntimeError: PubSub event loop is not set。4. 作用域hub 是进程内内存路由。每个 Flet 服务端进程TCP/Unix Socket 传输的FletSocketServer见 flet_socket_server.py桌面/移动端 Dart Bridge 传输的FletDartBridgeServer见 flet_dart_bridge_server.py在启动时都会创建PubSubHub(looploop, executorexecutor)。而 Pyodide 模式下pyodide_connection.py创建的是不带 loop/executor 的裸 hub。这意味着消息无法跨进程传递——若部署了多个服务进程如多 worker 的 Web 服务各进程间的 PubSub 是隔离的这一点在设计架构时需特别注意。会话生命周期与自动清理PubSubClient的订阅与会话生命周期强绑定无需担心僵尸订阅每个会话创建时自动获得自己的客户端session.py会话 ID 全局唯一会话关闭Session.close()时Flet 会自动调用self.__pubsub_client.unsubscribe_all()清掉该会话的全部订阅session.py因此即使开发者忘记在page.on_close里手动退订服务端会话回收时也会兜底清理不会把消息发给已断开的会话。不过官方教程仍然建议在page.on_close中显式调用unsubscribe_all()以尽早释放回调闭包引用的对象避免内存滞留。使用建议与边界全局广播 vs 主题所有会话都该收到的用send_all/subscribe只想让特定分组如房间 1的成员收到的用send_all_on_topic/subscribe_topic。主题名就是普通字符串由你自行约定命名规范如room:101。自己是否接收默认send_all会把消息发回发起者自己若回调逻辑会把消息追加到 UI建议用send_others避免重复显示或让回调按会话判断来源。处理器重入同一会话重复subscribe同一 handler 时由于 handler 存放在set中重复注册会被自动去重pubsub_hub.py无需担心重复回调。进程内限定PubSub 不跨进程、不跨机器属于轻量级内存消息总线需要跨服务实例的分布式消息应引入独立的消息中间件。两个类均从flet顶层包直接导出PubSubClient、PubSubHub见init.py模块级导出定义在 pubsub/init.py其中 hub 的完整 API 可参考 PubSubHub 文档。掌握PubSubClient你就掌握了 Flet 多用户实时应用的消息骨架。赞分享前端跨平台桌面应用移动开发【免费下载链接】fletBuild realtime web, mobile and desktop apps in Python only. No frontend experience required.项目地址https://gitcode.com/gh_mirrors/fl/flet点击查看免费下载相关推荐PubSubClient消息处理深度解析发布与订阅的完整流程PubSubClient消息处理深度解析发布与订阅的完整流程 PubSubClient是一个专为Arduino平台设计的MQTT客户端库它提供了完整的发布订物联网消息队列Azure Service Bus Python SDK 实战指南队列、主题订阅与会话消息全解析Azure Service Bus Python SDK 实战指南队列、主题订阅与会话消息全解析 本指南基于 agentic awesome skills 仓AI 技能AI 插件AnotherRedisDesktopManager 发布订阅功能实时消息接收与发送AnotherRedisDesktopManager 发布订阅功能实时消息接收与发送 引言Redis 发布订阅的应用场景与痛点 你是否在分布式系统开发中遇到数据库客户端桌面应用上一篇Manim Community Edition 文档导航与快速上手从安装到第一个数学动画下一篇Cloudflare Cache Reserve API 实战指南Workers 集成、缓存清理与监控分析创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考