当前位置：首页 > news >正文

12 极物科技 JetLinks MQTT直连设备事件上报实战（继电器场景）

news 2026/5/23 5:38:06

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）保证消息传输，应用层无需平台返回“上报成功/失败”，设备只需按规范发送即可。

（2）data字段和物模型怎么对应？

data内的字段需和物模型中事件的输出参数完全匹配（字段名、数据类型一致），对应关系看下图一目了然：

图1：物模型事件输出参数与报文data字段对应关系

注意：data这几个字符串不要改哈，jetlink官方的文档有时候看起不不是特别直接。

3. 物模型事件创建

3.1 创建步骤

在JetLinks平台对应产品下创建事件，核心配置如下：

事件ID：switchchange（需和代码中EVENT_ID一致）
事件名称：继电器状态变更
输出参数：
- 参数名：switch2
- 数据类型：字符串
- 可选值：on（开启）、off（关闭）

3.2 概念解析

（1）输出参数的含义？

事件的“输出参数”是事件触发时携带的业务数据，用于描述事件的具体状态。比如本场景中，通过switch2的on/off区分继电器“开启”和“关闭”两种状态。

（2）明明属性也能传状态，为什么用事件？

很多新手会混淆“属性”和“事件”，其中最核心的差异是事件可触发规则引擎的复杂联动，其次是存储和语义的区别，完整对比如下：

维度	属性上报	事件上报
核心语义	设备当前的静态状态	设备发生的动态行为/动作
存储逻辑	覆盖式存储（仅保留最新值）	日志式存储（保留所有记录）
规则引擎支持	仅能触发简单的“属性变更”规则	可触发复杂联动（告警、设备控制、第三方接口调用等）
适用场景	实时查看当前状态（如“现在开关是开的”）	1. 追溯状态变更（如“什么时候从开变关”） 2. 触发业务联动（如开关关闭时推送告警、开启时控制其他设备）

继电器状态变更场景中，事件的核心价值并非仅记录历史，而是能基于“开关开启/关闭”这个行为触发规则引擎的复杂操作：比如开关关闭时推送短信告警、开关开启时联动其他设备启动，这些都是属性上报无法实现的核心能力。

（3）为什么不拆分成“开关开启/关闭”两个事件？

冗余：两个事件本质都是“状态变更”，仅参数值不同，拆分无意义；
维护成本高：代码和物模型需维护两个事件ID，增加复杂度；
规则适配难：后续配置告警/联动时，需写两条规则，不如单事件+参数区分高效。

4. 完整代码实现（可直接运行，问题奀）

importjsonimportloggingimportrandomimporttimeimportuuidfrompaho.mqttimportclientasmqtt_clientfrompaho.mqtt.clientimportMQTTv311# ===================== 配置项 ======================MQTT_BROKER="192.168.111.53"MQTT_PORT=1883MQTT_USERNAME="admin"MQTT_PASSWORD="admin"CLIENT_ID="2019722818928308224"# 设备基础信息PRODUCT_ID="2019049642691198976"DEVICE_ID="2019722818928308224"# 事件配置（核心：事件标识 switchchange ，上报间隔5秒）EVENT_ID="switchchange"# 事件标识REPORT_INTERVAL=5# 上报间隔：5秒EVENT_REPORT_TOPIC=f"/{PRODUCT_ID}/{DEVICE_ID}/event/{EVENT_ID}"# 事件上报Topic# ========== 继电器设备状态 ==========DEVICE_STATE={"switch_status":False,# false=关闭，true=开启（核心布尔状态）}# 日志配置logging.basicConfig(level=logging.INFO,format="%(asctime)s - %(levelname)s - %(message)s")logger=logging.getLogger(__name__)defconnect_mqtt()->mqtt_client.Client:"""MQTT连接逻辑（保留核心，移除功能调用相关）"""defon_connect(client,userdata,flags,rc,properties=None):rc_msg={0:"连接成功",1:"协议版本错误",2:"客户端ID非法",3:"服务器不可用",4:"用户名/密码错误",5:"未授权",}ifrc==0:logger.info(f"✅ MQTT连接成功（{MQTT_BROKER}:{MQTT_PORT}）")else:logger.error(f"❌ 连接失败（rc={rc}）：{rc_msg.get(rc,'未知错误')}")defon_disconnect(client,userdata,rc,properties=None):ifrc!=0:logger.warning(f"⚠️ MQTT被动断开，将自动重连（rc={rc}）")# 创建客户端client=mqtt_client.Client(client_id=CLIENT_ID,callback_api_version=mqtt_client.CallbackAPIVersion.VERSION1,protocol=MQTTv311,)client.username_pw_set(MQTT_USERNAME,MQTT_PASSWORD)client.on_connect=on_connect client.on_disconnect=on_disconnect client.auto_reconnect=Trueclient.reconnect_delay_set(min_delay=2,max_delay=10)# 连接服务器try:client.connect(MQTT_BROKER,MQTT_PORT,keepalive=60)exceptExceptionase:logger.error(f"❌ TCP连接失败：{str(e)}")raisereturnclientdefgenerate_event_payload()->str:"""构造事件上报报文（严格匹配格式：timestamp+messageId+data(布尔类型)）"""# ========== 关键修改：根据switch_status动态设置switch2的值 ==========switch2_value="on"ifDEVICE_STATE["switch_status"]else"off"payload=json.dumps({"data":{"switch":DEVICE_STATE["switch_status"],"switch2":switch2_value,# 动态赋值为on/off},},ensure_ascii=False,)returnpayloaddefupdate_switch_status():"""每5秒切换开关状态（true↔false）"""DEVICE_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. 生成事件上报报文payload=generate_event_payload()# 3. 发布事件消息publish_result=client.publish(EVENT_REPORT_TOPIC,payload,qos=0)ifpublish_result[0]==0:logger.info(f"✅ 事件上报成功：")logger.info(f" 📝 主题:{EVENT_REPORT_TOPIC}")logger.info(f" 📝 报文:{json.dumps(json.loads(payload),ensure_ascii=False,indent=2)}")else:logger.error(f"❌ 事件上报失败，状态码：{publish_result[0]}")# 4. 等待5秒后继续time.sleep(REPORT_INTERVAL)exceptKeyboardInterrupt:logger.info("\n🛑 停止事件上报服务")breakexceptExceptionase:logger.error(f"❌ 事件上报异常：{str(e)}",exc_info=True)time.sleep(REPORT_INTERVAL)# 异常后仍等待，避免频繁报错defrun():"""启动开关状态事件上报服务"""logger.info("🚀 启动JetLinks 开关状态MQTT事件上报服务")client=connect_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_info=True)