发表新帖
发表新帖

【设备端物模型历史数据上报】代码实现与技术手册中协议不一致，导致broker解析异常

IOT 未结
关注 举报
2 426
hanbd
hanbd 剑圣 2025-02-12 11:29

一、该问题的重现步骤是什么？

  1. 按 技术手册 -》序言 -》 协议介绍 -》6.6 设备端物模型历史数据上报 中描述的请求数据格式组织数据。

  2. 向对应Topic发送mqtt数据

  3. broker报解析异常


二、你期待的结果是什么？实际看到的又是什么？

期待结果：能够按照技术手册 -》序言 -》 协议介绍 -》6.6 设备端物模型历史数据上报 章节描述的消息体格式批量进行历史数据上报

实际看到的：

  1. 技术手册协议描述中，properties支持数组格式，对于单个设备来说，可以一次上传多条数据

  2. broker中代码实现中，properties为Map格式，与协议不一致，对应类


org.springblade.mqtt.broker.data.function.event.EventPropertyHistoryPostFunction
org.springblade.mqtt.broker.data.protocol.event.PropertyHistoryPostReq

image.png

image.png

三、你正在使用的是什么产品，什么版本？在什么操作系统上？

Blade-Links

v1.3.0

Windows 11


四、请提供详细的错误堆栈信息，这很重要。

示例数据：

{
	"id": "5d75530cc1dad31cec48e954ad93a92a",
	"version": "1.0",
	"sys": {
		"ack": 0
	},
	"method": "thing.event.property.history.post",
	"params": [{
		"identity": {
			"productKey": "ZT_slj",
			"deviceName": "jd_slj_1"
		},
		"properties": [{
			"cf": {
				"value": 1.094759,
				"time": 1739327857344
			}
		}, {
			"cf": {
				"value": 7.7139072,
				"time": 1739327862344
			}
		}, {
			"cf": {
				"value": 3.0228767,
				"time": 1739327867344
			}
		}]
	}]
}

broker异常信息：

2025-02-12 11:27:32.534 ERROR 22292 --- [  biz-worker-31] n.d.i.m.c.s.s.DefaultMqttServerProcessor : com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot deserialize value of type `java.util.LinkedHashMap<java.lang.String,java.lang.Object>` from Array value (token `JsonToken.START_ARRAY`)
 at [Source: (byte[])"{"id":"ffa6ad5ece6498cdddf5b4db1a7829ff","version":"1.0","sys":{"ack":0},"method":"thing.event.property.history.post","params":[{"identity":{"productKey":"ZT_slj","deviceName":"jd_slj_1"},"properties":[{"cf":{"value":8.746042,"time":1739330847926}},{"cf":{"value":-5.9267864,"time":1739330852926}},{"cf":{"value":5.870201,"time":1739330857926}}]}]}"; line: 1, column: 202] (through reference chain: org.springblade.mqtt.broker.data.protocol.core.DataReq["params"]->java.util.ArrayList[0]->org.springblade.mqtt.broker.data.protocol.event.PropertyHistoryPostReq["properties"])

java.lang.RuntimeException: com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot deserialize value of type `java.util.LinkedHashMap<java.lang.String,java.lang.Object>` from Array value (token `JsonToken.START_ARRAY`)
 at [Source: (byte[])"{"id":"ffa6ad5ece6498cdddf5b4db1a7829ff","version":"1.0","sys":{"ack":0},"method":"thing.event.property.history.post","params":[{"identity":{"productKey":"ZT_slj","deviceName":"jd_slj_1"},"properties":[{"cf":{"value":8.746042,"time":1739330847926}},{"cf":{"value":-5.9267864,"time":1739330852926}},{"cf":{"value":5.870201,"time":1739330857926}}]}]}"; line: 1, column: 202] (through reference chain: org.springblade.mqtt.broker.data.protocol.core.DataReq["params"]->java.util.ArrayList[0]->org.springblade.mqtt.broker.data.protocol.event.PropertyHistoryPostReq["properties"])
	at org.springblade.core.tool.utils.Exceptions.unchecked(Exceptions.java:46)
	at org.springblade.core.tool.jackson.JsonUtil.readValue(JsonUtil.java:422)
	at org.springblade.mqtt.broker.data.listener.BladeMqttFunctionMessageListener.execDeviceFunction(BladeMqttFunctionMessageListener.java:103)
	at org.springblade.mqtt.broker.data.listener.BladeMqttFunctionMessageListener.execMqttFunction(BladeMqttFunctionMessageListener.java:81)
	at org.springblade.mqtt.broker.data.listener.BladeMqttFunctionMessageListener.lambda$onMessage$0(BladeMqttFunctionMessageListener.java:69)
	at org.springblade.core.tenant.TenantUtil.use(TenantUtil.java:93)
	at org.springblade.mqtt.broker.data.listener.BladeMqttFunctionMessageListener.onMessage(BladeMqttFunctionMessageListener.java:68)
	at net.dreamlu.iot.mqtt.core.server.support.DefaultMqttServerProcessor.lambda$invokeListenerForPublish$4(DefaultMqttServerProcessor.java:514)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
	at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:842)
Caused by: com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot deserialize value of type `java.util.LinkedHashMap<java.lang.String,java.lang.Object>` from Array value (token `JsonToken.START_ARRAY`)
 at [Source: (byte[])"{"id":"ffa6ad5ece6498cdddf5b4db1a7829ff","version":"1.0","sys":{"ack":0},"method":"thing.event.property.history.post","params":[{"identity":{"productKey":"ZT_slj","deviceName":"jd_slj_1"},"properties":[{"cf":{"value":8.746042,"time":1739330847926}},{"cf":{"value":-5.9267864,"time":1739330852926}},{"cf":{"value":5.870201,"time":1739330857926}}]}]}"; line: 1, column: 202] (through reference chain: org.springblade.mqtt.broker.data.protocol.core.DataReq["params"]->java.util.ArrayList[0]->org.springblade.mqtt.broker.data.protocol.event.PropertyHistoryPostReq["properties"])
	at com.fasterxml.jackson.databind.exc.MismatchedInputException.from(MismatchedInputException.java:59)
	at com.fasterxml.jackson.databind.DeserializationContext.reportInputMismatch(DeserializationContext.java:1752)
	at com.fasterxml.jackson.databind.DeserializationContext.handleUnexpectedToken(DeserializationContext.java:1526)
	at com.fasterxml.jackson.databind.deser.std.StdDeserializer._deserializeFromArray(StdDeserializer.java:222)
	at com.fasterxml.jackson.databind.deser.std.MapDeserializer.deserialize(MapDeserializer.java:457)
	at com.fasterxml.jackson.databind.deser.std.MapDeserializer.deserialize(MapDeserializer.java:32)
	at com.fasterxml.jackson.databind.deser.impl.MethodProperty.deserializeAndSet(MethodProperty.java:129)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.vanillaDeserialize(BeanDeserializer.java:314)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:177)
	at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer._deserializeFromArray(CollectionDeserializer.java:359)
	at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer.deserialize(CollectionDeserializer.java:244)
	at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer.deserialize(CollectionDeserializer.java:28)
	at com.fasterxml.jackson.databind.deser.impl.MethodProperty.deserializeAndSet(MethodProperty.java:129)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.vanillaDeserialize(BeanDeserializer.java:314)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:177)
	at com.fasterxml.jackson.databind.deser.DefaultDeserializationContext.readRootValue(DefaultDeserializationContext.java:323)
	at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4825)
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3866)
	at org.springblade.core.tool.jackson.JsonUtil.readValue(JsonUtil.java:420)
	... 12 common frames omitted


五、若有更多详细信息，请在下面提供。

2条回答
  • hanbd
    2025-02-12 13:45

    我简单做了代码修改，临时能达到协议效果，供参考。共有两处修改：

    1. 修改类org.springblade.mqtt.broker.data.protocol.event.PropertyHistoryPostReq

    2. 修改方法org.springblade.mqtt.broker.data.function.event.EventPropertyHistoryPostFunction#saveHistoryData

    package org.springblade.mqtt.broker.data.protocol.event;

import lombok.Data;

import java.util.List;
import java.util.Map;

/**
 * 设备批量上报属性、事件
 *
 * @author L.cm
 */
@Data
public class PropertyHistoryPostReq {
    /**
     * 设备信息
     */
    private Map<String, String> identity;
    /**
     * 属性信息
     */
    private List<Map<String, Object>> properties;
    /**
     * 事件
     */
    private List<Map<String, Object>> events;
}
    /**
     * 保存数据
     *
     * @param params 历史列表
     */
    private void saveHistoryData(List<PropertyHistoryPostReq> params) {
        for (PropertyHistoryPostReq param : params) {
           Map<String, String> identity = param.getIdentity();
           // 设备属性
           List<Map<String, Object>> propertiesList = param.getProperties();
           if (CollectionUtil.isNotEmpty(propertiesList)) {
              propertiesList
                 .stream()
                 .filter(CollectionUtil::isNotEmpty)
                 .forEach(properties -> deviceDataService.saveDeviceProperty(identity, properties));
           }
           // 设备事件
           List<Map<String, Object>> eventsList = param.getEvents();
           if (CollectionUtil.isNotEmpty(eventsList)) {
              eventsList
                 .stream()
                 .filter(CollectionUtil::isNotEmpty)
                 .forEach(events -> deviceDataService.saveDeviceEvent(identity, events));
           }
        }
    }


    ---

    后续版本建议修复该问题

    0 讨论(0)
  • 如梦技术
    2025-02-12 13:46

    https://center.javablade.com/blade/BladeX-Links/commit/f01bd95b2ccf8b0068941251d7c01e073be542bd


    感谢反馈，master 上做了修复，先用 for 循环吧，下一个版本会优化！

    0 讨论(0)
 看不清?
提交回复
官方产品
BladeX 企业级开发平台 BladeX 可视化数据大屏
官方新品
BladeX 大模型开发平台 BladeX 物联网开发平台
热议问题
物联网我现在用用 MQTT + Kafka + Node-RED 这种模式， Node-RED数据清洗以后如何发送到平台数据库？ 1
bladex使用kingbase连接后，查询数据要在前面拼接`mysql`.请问如何解决？ 1
最新版本bladex拉取maven仓库失败 1
edis实现的接口防抖工具如何开启呢，目前不生效 2
4.6.0第三方登录，没有注册步骤 1
统一认证登录注销接口改进建议 1
服务启动报错获取不到Nacos的资源 1
我们需要修改工作流，框架里面的工作流没有开放源码么？框架里面的工作流是个包？ 1
后台重启后，导入提示令牌过期 1
通过第三方认证接口/oauth/render访问bladex的统一认证平台，无法通过/oauth/revoke注销 1
layui

扫一扫访问 Blade技术社区 移动端

Blade技术社区 2025 © 上海布雷德科技有限公司 沪ICP备2023009528号-1 苏公网安备 32041102000998号