JetLinks MQTT直连设备事件上报实战继电器场景1. 前言JetLinks作为国产开源物联网平台MQTT协议直连设备是最常用的接入方式之一。但很多开发者在实操中会遇到两个核心问题一是分不清“事件上报”和“属性上报”的使用场景二是JetLinks 2.10社区版存在“事件上报后无法直接在页面查看”的坑导致测试时无从验证本人在验证的时候折腾了半天最后通过事件触发告警(需要编写规则引擎)告警记录可查看来验证事件是否上报成功太难了坑爹也有可能是本人某些地方设置有问题但是使用当前文档的方法测试是完全没问题的。本文以继电器开关状态事件上报为例从协议规范、物模型配置、代码实现到测试验证完整拆解JetLinks MQTT直连设备的事件上报全流程重点解决“社区版怎么验证事件上报成功”的核心痛点内容可直接落地。突然想到可能和存储有关系开启TimescaleDB-单列模式,果然看到事件日志了,不过还是不影响本人的证明过程问题奀可防可控看第7小结如何设置即可愉快使用了2. 事件上报规范主题报文2.1 上报主题JetLinks对MQTT直连设备的事件上报主题做了固定约定格式如下/{productId}/{deviceId}/event/{eventId}传输方向上行设备 → 平台字段说明productId平台创建产品时生成的产品唯一标识deviceId产品下创建设备时生成的设备唯一标识eventId物模型中定义的事件ID需和配置完全一致2.2 报文格式事件上报报文无需冗余字段仅需传输核心业务数据即可平台会自动补全时间戳、messageID等元数据{data:{switch2:on}}2.3 核心疑问解答1设备需要获取上报返回值吗不需要。MQTT协议层面可通过QoS本文用QoS0追求可靠性可改QoS1保证消息传输应用层无需平台返回“上报成功/失败”设备只需按规范发送即可。2data字段和物模型怎么对应data内的字段需和物模型中事件的输出参数完全匹配字段名、数据类型一致对应关系看下图一目了然图1物模型事件输出参数与报文data字段对应关系注意data这几个字符串不要改哈jetlink官方的文档有时候看起不不是特别直接。3. 物模型事件创建3.1 创建步骤在JetLinks平台对应产品下创建事件核心配置如下事件IDswitchchange需和代码中EVENT_ID一致事件名称继电器状态变更输出参数参数名switch2数据类型字符串可选值on开启、off关闭3.2 概念解析1输出参数的含义事件的“输出参数”是事件触发时携带的业务数据用于描述事件的具体状态。比如本场景中通过switch2的on/off区分继电器“开启”和“关闭”两种状态。2明明属性也能传状态为什么用事件很多新手会混淆“属性”和“事件”其中最核心的差异是事件可触发规则引擎的复杂联动其次是存储和语义的区别完整对比如下维度属性上报事件上报核心语义设备当前的静态状态设备发生的动态行为/动作存储逻辑覆盖式存储仅保留最新值日志式存储保留所有记录规则引擎支持仅能触发简单的“属性变更”规则可触发复杂联动告警、设备控制、第三方接口调用等适用场景实时查看当前状态如“现在开关是开的”1. 追溯状态变更如“什么时候从开变关”2. 触发业务联动如开关关闭时推送告警、开启时控制其他设备继电器状态变更场景中事件的核心价值并非仅记录历史而是能基于“开关开启/关闭”这个行为触发规则引擎的复杂操作比如开关关闭时推送短信告警、开关开启时联动其他设备启动这些都是属性上报无法实现的核心能力。3为什么不拆分成“开关开启/关闭”两个事件冗余两个事件本质都是“状态变更”仅参数值不同拆分无意义维护成本高代码和物模型需维护两个事件ID增加复杂度规则适配难后续配置告警/联动时需写两条规则不如单事件参数区分高效。4. 完整代码实现可直接运行问题奀importjsonimportloggingimportrandomimporttimeimportuuidfrompaho.mqttimportclientasmqtt_clientfrompaho.mqtt.clientimportMQTTv311# 配置项 MQTT_BROKER192.168.111.53MQTT_PORT1883MQTT_USERNAMEadminMQTT_PASSWORDadminCLIENT_ID2019722818928308224# 设备基础信息PRODUCT_ID2019049642691198976DEVICE_ID2019722818928308224# 事件配置核心事件标识 switchchange 上报间隔5秒EVENT_IDswitchchange# 事件标识REPORT_INTERVAL5# 上报间隔5秒EVENT_REPORT_TOPICf/{PRODUCT_ID}/{DEVICE_ID}/event/{EVENT_ID}# 事件上报Topic# 继电器设备状态 DEVICE_STATE{switch_status:False,# false关闭true开启核心布尔状态}# 日志配置logging.basicConfig(levellogging.INFO,format%(asctime)s - %(levelname)s - %(message)s)loggerlogging.getLogger(__name__)defconnect_mqtt()-mqtt_client.Client:MQTT连接逻辑保留核心移除功能调用相关defon_connect(client,userdata,flags,rc,propertiesNone):rc_msg{0:连接成功,1:协议版本错误,2:客户端ID非法,3:服务器不可用,4:用户名/密码错误,5:未授权,}ifrc0:logger.info(f✅ MQTT连接成功{MQTT_BROKER}:{MQTT_PORT})else:logger.error(f❌ 连接失败rc{rc}{rc_msg.get(rc,未知错误)})defon_disconnect(client,userdata,rc,propertiesNone):ifrc!0:logger.warning(f⚠️ MQTT被动断开将自动重连rc{rc})# 创建客户端clientmqtt_client.Client(client_idCLIENT_ID,callback_api_versionmqtt_client.CallbackAPIVersion.VERSION1,protocolMQTTv311,)client.username_pw_set(MQTT_USERNAME,MQTT_PASSWORD)client.on_connecton_connect client.on_disconnecton_disconnect client.auto_reconnectTrueclient.reconnect_delay_set(min_delay2,max_delay10)# 连接服务器try:client.connect(MQTT_BROKER,MQTT_PORT,keepalive60)exceptExceptionase:logger.error(f❌ TCP连接失败{str(e)})raisereturnclientdefgenerate_event_payload()-str:构造事件上报报文严格匹配格式timestampmessageIddata(布尔类型)# 关键修改根据switch_status动态设置switch2的值 switch2_valueonifDEVICE_STATE[switch_status]elseoffpayloadjson.dumps({data:{switch:DEVICE_STATE[switch_status],switch2:switch2_value,# 动态赋值为on/off},},ensure_asciiFalse,)returnpayloaddefupdate_switch_status():每5秒切换开关状态true↔falseDEVICE_STATE[switch_status]notDEVICE_STATE[switch_status]status_text开启ifDEVICE_STATE[switch_status]else关闭logger.info(f 开关状态已更新{status_text}{DEVICE_STATE[switch_status]})defreport_event_periodically(client:mqtt_client.Client):周期性上报事件每5秒一次logger.info(f 开始周期性上报事件间隔{REPORT_INTERVAL}秒)logger.info(f 事件标识{EVENT_ID})logger.info(f 事件上报Topic{EVENT_REPORT_TOPIC})logger.info(f 初始开关状态{关闭ifnotDEVICE_STATE[switch_status]else开启})whileTrue:try:# 1. 切换开关状态update_switch_status()# 2. 生成事件上报报文payloadgenerate_event_payload()# 3. 发布事件消息publish_resultclient.publish(EVENT_REPORT_TOPIC,payload,qos0)ifpublish_result[0]0:logger.info(f✅ 事件上报成功)logger.info(f 主题:{EVENT_REPORT_TOPIC})logger.info(f 报文:{json.dumps(json.loads(payload),ensure_asciiFalse,indent2)})else:logger.error(f❌ 事件上报失败状态码{publish_result[0]})# 4. 等待5秒后继续time.sleep(REPORT_INTERVAL)exceptKeyboardInterrupt:logger.info(\n 停止事件上报服务)breakexceptExceptionase:logger.error(f❌ 事件上报异常{str(e)},exc_infoTrue)time.sleep(REPORT_INTERVAL)# 异常后仍等待避免频繁报错defrun():启动开关状态事件上报服务logger.info( 启动JetLinks 开关状态MQTT事件上报服务)clientconnect_mqtt()client.loop_start()# 启动非阻塞MQTT循环try:report_event_periodically(client)finally:client.loop_stop()client.disconnect()logger.info( MQTT连接已断开)if__name____main__:try:run()exceptExceptionase:logger.error(f❌ 程序异常退出{str(e)},exc_infoTrue)4.3 代码说明配置项需替换为自己平台的productId、deviceId、MQTT地址等信息报文优化移除了多余的switch字段仅保留物模型定义的switch2严格匹配平台规范异常处理包含连接失败、上报异常、用户终止等场景的友好提示便于调试。代码都在本人的gitee上,代码无保留热心群众https://gitee.com/fujianxinxi/tearcher.git5. 测试验证社区版“曲线救国”方案5.1 测试背景JetLinks 2.10社区版存在限制事件上报后无法直接在“设备事件”页面看到记录需通过规则引擎告警记录验证有点麻烦就是了不过能用就行先用再说。5.2 测试步骤步骤1配置规则引擎事件触发告警进入平台「规则引擎」→ 新建规则触发条件选择“设备事件”→ 关联对应产品事件IDswitchchange执行动作选择“发送告警”→ 配置告警标题如“继电器状态变更”启用规则。正常运行日志如下图2Python代码运行日志—— 显示周期性上报on/off报文步骤3查看告警记录验证进入平台「告警中心」→「告警记录」可看到继电器状态变更的告警记录图3Web端告警记录—— 事件触发的告警信息5.3 验证结论告警记录的触发时间、switch2参数值和代码上报的内容完全一致证明事件已成功上报到平台。务必注意不是代码发送有问题是社区版本的功能限制导致无法直接看事件记录6. 踩坑记录社区版必看坑点1社区版事件页面无记录最坑图4社区版事件页面无记录—— 2.10版本社区版事件页面为空现象事件上报后「运行状态 tab」页面始终为空,上图所示原因JetLinks 2.10社区版未开放事件日志的直接展示功能解决方案通过规则引擎关联告警用告警记录验证事件上报结果曲线救国没招只能这样。坑点2规则引擎配置事件关联告警图5规则引擎配置—— 事件关联告警的规则配置关键触发条件必须精准选择“对应产品事件ID”否则告警无法触发提示规则配置完成后需“启用”否则不生效。坑点3最终验证告警记录图6告警记录页面—— 最终验证事件上报的告警记录核心告警记录的“触发时间”“参数值”需和代码上报的一致才算验证通过吐槽这种验证方式确实反人类但社区版只能这么玩7. 尴尬突然想到可能和存储有关系开启TimescaleDB-单列模式,果然看到事件日志了.8. 结论JetLinks事件上报核心主题固定、报文极简仅保留物模型定义的data字段无需额外元数据事件vs属性状态变更用事件追溯历史实时状态用属性查看当前关联场景编辑复杂规则引擎社区版验证技巧事件无直接展示时通过“规则引擎告警记录”曲线验证这是2.10版本的可行方案避坑关键代码配置项需和平台物模型/设备信息严格一致报文字段不能多也不能少。本文完整覆盖了JetLinks MQTT设备事件上报的全流程解决了新手最易踩的“配置乱、验证难”问题。如果有其他JetLinks使用问题欢迎在评论区交流
12 极物科技 JetLinks MQTT直连设备事件上报实战（继电器场景）
发布时间：2026/5/21 3:59:23
JetLinks MQTT直连设备事件上报实战继电器场景1. 前言JetLinks作为国产开源物联网平台MQTT协议直连设备是最常用的接入方式之一。但很多开发者在实操中会遇到两个核心问题一是分不清“事件上报”和“属性上报”的使用场景二是JetLinks 2.10社区版存在“事件上报后无法直接在页面查看”的坑导致测试时无从验证本人在验证的时候折腾了半天最后通过事件触发告警(需要编写规则引擎)告警记录可查看来验证事件是否上报成功太难了坑爹也有可能是本人某些地方设置有问题但是使用当前文档的方法测试是完全没问题的。本文以继电器开关状态事件上报为例从协议规范、物模型配置、代码实现到测试验证完整拆解JetLinks MQTT直连设备的事件上报全流程重点解决“社区版怎么验证事件上报成功”的核心痛点内容可直接落地。突然想到可能和存储有关系开启TimescaleDB-单列模式,果然看到事件日志了,不过还是不影响本人的证明过程问题奀可防可控看第7小结如何设置即可愉快使用了2. 事件上报规范主题报文2.1 上报主题JetLinks对MQTT直连设备的事件上报主题做了固定约定格式如下/{productId}/{deviceId}/event/{eventId}传输方向上行设备 → 平台字段说明productId平台创建产品时生成的产品唯一标识deviceId产品下创建设备时生成的设备唯一标识eventId物模型中定义的事件ID需和配置完全一致2.2 报文格式事件上报报文无需冗余字段仅需传输核心业务数据即可平台会自动补全时间戳、messageID等元数据{data:{switch2:on}}2.3 核心疑问解答1设备需要获取上报返回值吗不需要。MQTT协议层面可通过QoS本文用QoS0追求可靠性可改QoS1保证消息传输应用层无需平台返回“上报成功/失败”设备只需按规范发送即可。2data字段和物模型怎么对应data内的字段需和物模型中事件的输出参数完全匹配字段名、数据类型一致对应关系看下图一目了然图1物模型事件输出参数与报文data字段对应关系注意data这几个字符串不要改哈jetlink官方的文档有时候看起不不是特别直接。3. 物模型事件创建3.1 创建步骤在JetLinks平台对应产品下创建事件核心配置如下事件IDswitchchange需和代码中EVENT_ID一致事件名称继电器状态变更输出参数参数名switch2数据类型字符串可选值on开启、off关闭3.2 概念解析1输出参数的含义事件的“输出参数”是事件触发时携带的业务数据用于描述事件的具体状态。比如本场景中通过switch2的on/off区分继电器“开启”和“关闭”两种状态。2明明属性也能传状态为什么用事件很多新手会混淆“属性”和“事件”其中最核心的差异是事件可触发规则引擎的复杂联动其次是存储和语义的区别完整对比如下维度属性上报事件上报核心语义设备当前的静态状态设备发生的动态行为/动作存储逻辑覆盖式存储仅保留最新值日志式存储保留所有记录规则引擎支持仅能触发简单的“属性变更”规则可触发复杂联动告警、设备控制、第三方接口调用等适用场景实时查看当前状态如“现在开关是开的”1. 追溯状态变更如“什么时候从开变关”2. 触发业务联动如开关关闭时推送告警、开启时控制其他设备继电器状态变更场景中事件的核心价值并非仅记录历史而是能基于“开关开启/关闭”这个行为触发规则引擎的复杂操作比如开关关闭时推送短信告警、开关开启时联动其他设备启动这些都是属性上报无法实现的核心能力。3为什么不拆分成“开关开启/关闭”两个事件冗余两个事件本质都是“状态变更”仅参数值不同拆分无意义维护成本高代码和物模型需维护两个事件ID增加复杂度规则适配难后续配置告警/联动时需写两条规则不如单事件参数区分高效。4. 完整代码实现可直接运行问题奀importjsonimportloggingimportrandomimporttimeimportuuidfrompaho.mqttimportclientasmqtt_clientfrompaho.mqtt.clientimportMQTTv311# 配置项 MQTT_BROKER192.168.111.53MQTT_PORT1883MQTT_USERNAMEadminMQTT_PASSWORDadminCLIENT_ID2019722818928308224# 设备基础信息PRODUCT_ID2019049642691198976DEVICE_ID2019722818928308224# 事件配置核心事件标识 switchchange 上报间隔5秒EVENT_IDswitchchange# 事件标识REPORT_INTERVAL5# 上报间隔5秒EVENT_REPORT_TOPICf/{PRODUCT_ID}/{DEVICE_ID}/event/{EVENT_ID}# 事件上报Topic# 继电器设备状态 DEVICE_STATE{switch_status:False,# false关闭true开启核心布尔状态}# 日志配置logging.basicConfig(levellogging.INFO,format%(asctime)s - %(levelname)s - %(message)s)loggerlogging.getLogger(__name__)defconnect_mqtt()-mqtt_client.Client:MQTT连接逻辑保留核心移除功能调用相关defon_connect(client,userdata,flags,rc,propertiesNone):rc_msg{0:连接成功,1:协议版本错误,2:客户端ID非法,3:服务器不可用,4:用户名/密码错误,5:未授权,}ifrc0:logger.info(f✅ MQTT连接成功{MQTT_BROKER}:{MQTT_PORT})else:logger.error(f❌ 连接失败rc{rc}{rc_msg.get(rc,未知错误)})defon_disconnect(client,userdata,rc,propertiesNone):ifrc!0:logger.warning(f⚠️ MQTT被动断开将自动重连rc{rc})# 创建客户端clientmqtt_client.Client(client_idCLIENT_ID,callback_api_versionmqtt_client.CallbackAPIVersion.VERSION1,protocolMQTTv311,)client.username_pw_set(MQTT_USERNAME,MQTT_PASSWORD)client.on_connecton_connect client.on_disconnecton_disconnect client.auto_reconnectTrueclient.reconnect_delay_set(min_delay2,max_delay10)# 连接服务器try:client.connect(MQTT_BROKER,MQTT_PORT,keepalive60)exceptExceptionase:logger.error(f❌ TCP连接失败{str(e)})raisereturnclientdefgenerate_event_payload()-str:构造事件上报报文严格匹配格式timestampmessageIddata(布尔类型)# 关键修改根据switch_status动态设置switch2的值 switch2_valueonifDEVICE_STATE[switch_status]elseoffpayloadjson.dumps({data:{switch:DEVICE_STATE[switch_status],switch2:switch2_value,# 动态赋值为on/off},},ensure_asciiFalse,)returnpayloaddefupdate_switch_status():每5秒切换开关状态true↔falseDEVICE_STATE[switch_status]notDEVICE_STATE[switch_status]status_text开启ifDEVICE_STATE[switch_status]else关闭logger.info(f 开关状态已更新{status_text}{DEVICE_STATE[switch_status]})defreport_event_periodically(client:mqtt_client.Client):周期性上报事件每5秒一次logger.info(f 开始周期性上报事件间隔{REPORT_INTERVAL}秒)logger.info(f 事件标识{EVENT_ID})logger.info(f 事件上报Topic{EVENT_REPORT_TOPIC})logger.info(f 初始开关状态{关闭ifnotDEVICE_STATE[switch_status]else开启})whileTrue:try:# 1. 切换开关状态update_switch_status()# 2. 生成事件上报报文payloadgenerate_event_payload()# 3. 发布事件消息publish_resultclient.publish(EVENT_REPORT_TOPIC,payload,qos0)ifpublish_result[0]0:logger.info(f✅ 事件上报成功)logger.info(f 主题:{EVENT_REPORT_TOPIC})logger.info(f 报文:{json.dumps(json.loads(payload),ensure_asciiFalse,indent2)})else:logger.error(f❌ 事件上报失败状态码{publish_result[0]})# 4. 等待5秒后继续time.sleep(REPORT_INTERVAL)exceptKeyboardInterrupt:logger.info(\n 停止事件上报服务)breakexceptExceptionase:logger.error(f❌ 事件上报异常{str(e)},exc_infoTrue)time.sleep(REPORT_INTERVAL)# 异常后仍等待避免频繁报错defrun():启动开关状态事件上报服务logger.info( 启动JetLinks 开关状态MQTT事件上报服务)clientconnect_mqtt()client.loop_start()# 启动非阻塞MQTT循环try:report_event_periodically(client)finally:client.loop_stop()client.disconnect()logger.info( MQTT连接已断开)if__name____main__:try:run()exceptExceptionase:logger.error(f❌ 程序异常退出{str(e)},exc_infoTrue)4.3 代码说明配置项需替换为自己平台的productId、deviceId、MQTT地址等信息报文优化移除了多余的switch字段仅保留物模型定义的switch2严格匹配平台规范异常处理包含连接失败、上报异常、用户终止等场景的友好提示便于调试。代码都在本人的gitee上,代码无保留热心群众https://gitee.com/fujianxinxi/tearcher.git5. 测试验证社区版“曲线救国”方案5.1 测试背景JetLinks 2.10社区版存在限制事件上报后无法直接在“设备事件”页面看到记录需通过规则引擎告警记录验证有点麻烦就是了不过能用就行先用再说。5.2 测试步骤步骤1配置规则引擎事件触发告警进入平台「规则引擎」→ 新建规则触发条件选择“设备事件”→ 关联对应产品事件IDswitchchange执行动作选择“发送告警”→ 配置告警标题如“继电器状态变更”启用规则。正常运行日志如下图2Python代码运行日志—— 显示周期性上报on/off报文步骤3查看告警记录验证进入平台「告警中心」→「告警记录」可看到继电器状态变更的告警记录图3Web端告警记录—— 事件触发的告警信息5.3 验证结论告警记录的触发时间、switch2参数值和代码上报的内容完全一致证明事件已成功上报到平台。务必注意不是代码发送有问题是社区版本的功能限制导致无法直接看事件记录6. 踩坑记录社区版必看坑点1社区版事件页面无记录最坑图4社区版事件页面无记录—— 2.10版本社区版事件页面为空现象事件上报后「运行状态 tab」页面始终为空,上图所示原因JetLinks 2.10社区版未开放事件日志的直接展示功能解决方案通过规则引擎关联告警用告警记录验证事件上报结果曲线救国没招只能这样。坑点2规则引擎配置事件关联告警图5规则引擎配置—— 事件关联告警的规则配置关键触发条件必须精准选择“对应产品事件ID”否则告警无法触发提示规则配置完成后需“启用”否则不生效。坑点3最终验证告警记录图6告警记录页面—— 最终验证事件上报的告警记录核心告警记录的“触发时间”“参数值”需和代码上报的一致才算验证通过吐槽这种验证方式确实反人类但社区版只能这么玩7. 尴尬突然想到可能和存储有关系开启TimescaleDB-单列模式,果然看到事件日志了.8. 结论JetLinks事件上报核心主题固定、报文极简仅保留物模型定义的data字段无需额外元数据事件vs属性状态变更用事件追溯历史实时状态用属性查看当前关联场景编辑复杂规则引擎社区版验证技巧事件无直接展示时通过“规则引擎告警记录”曲线验证这是2.10版本的可行方案避坑关键代码配置项需和平台物模型/设备信息严格一致报文字段不能多也不能少。本文完整覆盖了JetLinks MQTT设备事件上报的全流程解决了新手最易踩的“配置乱、验证难”问题。如果有其他JetLinks使用问题欢迎在评论区交流
排查思路和方法)
