MQTT 消费者
中心服务器 MQTT 消费面全量清单（device-service 消费者 22 类 + MQTT 面监听器 6 行（每 destination 一行）+ main-service 1，共 29 行）。MQTT 消息链路：device-service MqttConfig 以 Paho v5 订阅 MQTT topic（QoS 2），按通道契约路由到内部 JMS destination，再由下列类处理；「MQTT 面向」列区分直通 topic（topic 即 destination）、谓词族汇入（源 topic 匹配谓词后汇入常量 destination）、$SYS broker 事件与内部 JMS（消息不来自 MQTT）。收录边界：MQTT 消费链路（consumer/ 目录 + MQTT 面监听器——janus 路由 / ptt janus 转发 / 回环归档 / $SYS 事件）；非 MQTT 的 JMS 内部 destination（core 契约常量 84 个中的内部部分）与其它事件监听器不入本表。MQTT topic 维度的全量清单见 MQTT 主题清单。
源码归纳 @ 2026-09-14（日期基线）。
本页为生成物，不可手改，重摸底刷新（重跑生成器整体覆盖）。
device-service
| 消费者类 | 监听 destination | MQTT 面向 | 职责（源码归纳） |
|---|---|---|---|
DataAcquisitionServerIpSyncConsumer |
data acquisition server ip sync |
内部 JMS（非 MQTT） | 采集服务器信息同步：按采集服务器遍历关联采集设备，比对在线状态并按需推送 DMS 载荷 |
IotV1Consumer |
— |
基类（无监听） | iot v1 消费者家族基类：统一解包 MQTT 请求信封（id/token/params/reply）、记录在线状态并按 reply 主题回包；自身无监听 |
IotV1ForwardJanusAdminToConsumer（extends IotV1ForwardJanusConsumer） |
server/center-server/iot/v1/forward/janus/admin/to |
MQTT 直通 | iot 通道 Janus 管理转发：按 transaction 路由并转发 janus 管理请求（管理凭据由服务端侧处理，调用方无需携带） |
IotV1ForwardJanusConsumer |
— |
基类（无监听） | Janus 转发消费者基类：统一转发与回环防护；自身无监听 |
IotV1ForwardJanusToConsumer（extends IotV1ForwardJanusConsumer） |
server/center-server/iot/v1/forward/janus/to |
MQTT 直通 | iot 通道 Janus 信令转发：按 transaction 路由并转发 janus 请求到目标 Janus 服务器 |
IotV1KeepAliveOnlineStatusConsumer（extends IotV1Consumer） |
server/center-server/iot/v1/keep-alive-online-status |
MQTT 直通 | iot v1 在线状态保活：记录客户端在线状态与最后在线时间（刷新在线状态缓存），无业务回包 |
IotV1ListCategoryInfoConsumer（extends IotV1Consumer） |
server/center-server/iot/v1/list-category-info |
MQTT 直通 | iot v1 分类信息查询：按设备编号与分类类别返回上报的分类信息列表 |
IotV1ListIceServerConsumer（extends IotV1Consumer） |
server/center-server/iot/v1/list-ice-server |
MQTT 直通 | iot v1 ICE 服务器查询：返回 WebRTC 打洞用的 ICE 服务器配置列表 |
IotV1ListOnlineStatusConsumer（extends IotV1Consumer） |
server/center-server/iot/v1/list-online-status |
MQTT 直通 | iot v1 在线状态查询：按 MQTT 客户端 ID 前缀批量返回在线状态 |
IotV1ReportCategoryInfoConsumer（extends IotV1Consumer） |
server/center-server/iot/v1/report-category-info |
MQTT 直通 | iot v1 分类信息上报：接收设备上报的分类统计数据并入库 |
IotV1RetrieveConsumer（extends IotV1Consumer） |
server/center-server/iot/v1/retrieve |
MQTT 直通 + 谓词族汇入 | iot v1 数据检索：按 params.items 检索词汇表项（ice / firmware / dms / ptt / upload/*，未知项静默忽略）并按 reply 主题回包 |
IotV1TelemetryConsumer（extends IotV1Consumer） |
server/center-server/iot/v1/telemetry |
MQTT 直通 + 谓词族汇入 | iot v1 遥测上报：接收设备遥测数据并按类别入库（信封 sub 区分 status / data 模式） |
JanusInboundRouter |
server center janus inbound |
MQTT 谓词族归档（janus 入站） | janus 回包路由：将 Janus 服务器回包按请求 transaction 路由回发起方 |
MeilisearchHistoryRetentionConsumer |
— |
非监听（定时任务调用） | 历史索引保留：按索引族独立门限周期清理 6 个月前历史索引 |
MeilisearchSpoolReplayConsumer |
— |
非监听（定时任务调用） | 历史数据缓冲重放：重读待写入缓冲并提交历史索引 |
MeilisearchSpoolWriteConsumer |
— |
非监听（定时任务调用） | 历史数据缓冲写：文档先落缓冲再入历史索引（含磁盘余量防护） |
NotifyToRefreshTheDeviceListConsumer |
notify to refresh the device list |
内部 JMS（非 MQTT） | 设备列表刷新通知：接收设备变更通知并通过 MQTT 向相关用户推送设备列表刷新 |
PttJanusForwardListener |
server/center-server/ptt/v1/forward/janus/admin/to |
MQTT 直通 | ptt 通道 Janus 管理转发：将对讲管理请求转发至目标 Janus 服务器（管理凭据由服务端侧处理） |
PttJanusForwardListener |
server/center-server/ptt/v1/forward/janus/to |
MQTT 直通 | ptt 通道 Janus 信令转发：将对讲信令请求转发至对讲组音频房间所在的 Janus 服务器 |
PttV1RetrieveConsumer（extends IotV1Consumer） |
server/center-server/ptt/v1/retrieve |
谓词族汇入 | ptt v1 数据检索：按 items 检索词汇表处理 ptt 通道检索请求并回包 |
PttV1TelemetryConsumer（extends IotV1Consumer） |
server/center-server/ptt/v1/telemetry |
谓词族汇入 | ptt v1 遥测上报：接收 ptt 通道遥测数据（如设备 PTT 程序版本）并入库 |
SysBrokersClientActivityListener |
sys-brokers-client-activity |
$SYS broker 事件（谓词族归档） | 客户端订阅活动记录：按 broker 订阅 / 退订事件更新订阅状态记录 |
SysBrokersClientLifecycleListener |
sys-brokers-client-lifecycle |
$SYS broker 事件（谓词族归档） | 客户端连接状态维护：按 broker 连接 / 断开事件更新客户端在线状态 |
TalkGroupRoomResultListener |
server/center-server/ptt/v1/forward/janus/admin/from/audiobridge/room |
MQTT 直通（内部回环） | 对讲组房间结果处理：按房间处理结果维护对讲组房间状态（内部回环通道，对接方无需对接） |
WebrtcV1ForwardJanusAdminToConsumer（extends IotV1ForwardJanusConsumer） |
server/center-server/webrtc/v1/forward/janus/admin/to |
MQTT 直通 | webrtc 通道 Janus 管理转发：转发 janus 管理请求（管理凭据由服务端侧处理，调用方无需携带） |
WebrtcV1ForwardJanusToConsumer（extends IotV1ForwardJanusConsumer） |
server/center-server/webrtc/v1/forward/janus/to |
MQTT 直通 | webrtc 通道 Janus 信令转发：按 transaction 路由并转发 janus 请求到目标 Janus 服务器 |
WebrtcV1RetrieveConsumer（extends IotV1Consumer） |
server/center-server/webrtc/v1/retrieve |
谓词族汇入 | webrtc v1 数据检索：按 items 检索词汇表处理 webrtc 通道检索请求并回包 |
WebrtcV1TelemetryConsumer（extends IotV1Consumer） |
server/center-server/webrtc/v1/telemetry |
谓词族汇入 | webrtc v1 遥测上报：接收 webrtc 通道遥测数据并入库 |
main-service
| 消费者类 | 监听 destination | MQTT 面向 | 职责（源码归纳） |
|---|---|---|---|
MqttAccountRemoveConsumer |
— |
函数式 bean（JMS 监听器调用） | MQTT 账号缓存清理：账号删除时清理 MQTT 用户 / ACL 相关缓存（由账号与用户变更消息触发） |
目录