在 TDengine 里 DESCRIBE 一张 SagooIoT 建出来的超级表，会看到这样的字段序列：
taos> DESCRIBE product_air_sensor;
field | type | length | note |
================================================================
ts | TIMESTAMP | 8 | |
p_temperature | FLOAT | 4 | |
p_temperature_time | TIMESTAMP | 8 | |
p_humidity | FLOAT | 4 | |
p_humidity_time| TIMESTAMP | 8 | |
p_switch | BOOL | 1 | |
p_switch_time | TIMESTAMP | 8 | |
device | VARCHAR | 255 | TAG |
这张表不是谁手工建的，也不是在建库脚本里写死的。它上面每一个 p_ 开头的列、每一个跟着 _time 的伴生列，都是从产品物模型里算出来的。产品里加一个属性，表里就多两列；改一个属性的类型，下次建表就用新类型。
这篇文章讲的就是这条链路：一份物模型 JSON，如何同时决定时序库的表结构、上报报文的解析结果、以及最终写进库里的值长什么样。文中的实现细节来自本地专业版发行版的源码，涉及 internal/logic/product、internal/model、internal/queues、pkg/tsd、network/core/logic/model/up/property 几个目录；产品概念与使用方式可对照文档站 https://iotdoc.sagoo.cn ，社区版源码在 https://github.com/orgs/sagoo-cloud/repositories 。
---
一、物模型是产品级的一份 schema
SagooIoT 的物模型（TSL）不是一个给前端画页面用的表单，它是产品维度的数据结构定义，四段成员：
// internal/model/dev_tsl.go
type TSL struct {
Key string `json:"key"` // 产品标识
Name string `json:"name"` // 产品名称
Properties []TSLProperty `json:"properties"` // 属性
Functions []TSLFunction `json:"functions"` // 功能
Events []TSLEvent `json:"events"` // 事件
Tags []TSLTag `json:"tags"` // 标签
}
四类成员各管一件事：
| 成员 | 方向 | 典型用途 | 是否落成表结构 |
|---|---|---|---|
| 属性 Properties | 设备 → 平台 | 温度、湿度、开关状态 | 是，每个属性两列 |
| 功能 Functions | 平台 → 设备 | 下发指令、设置参数 | 否，走服务调用 |
| 事件 Events | 设备 → 平台 | 告警、故障、按键 | 否，走告警与消息 |
| 标签 Tags | 静态描述 | 安装位置、批次、型号 | 是，落成 TAG 列 |
属性本身也只有五个字段，但每个字段都有约束：
type TSLProperty struct {
Key string `json:"key" v:"required|regex:^[A-Za-z_]+[\\w]*$#请输入属性标识|标识由字母、数字和下划线组成,且不能以数字开头"`
Name string `json:"name" v:"required#请输入属性名称"`
AccessMode int `json:"accessMode" v:"required#请选择是否只读"` // 0=读写,1=只读
ValueType TSLValueType `json:"valueType" v:"属性值"`
Desc string `json:"desc" v:"描述"`
}
这里的 regex 校验值得单独说一句。属性标识被限制成「字母或下划线开头，后面只能是字母数字下划线」，原因很实际：这个标识最终会变成数据库列名，而 TDengine 的列名不接受数字开头、不接受中划线、不接受空格。把约束放在录入界面而不是建表那一刻，用户在产品配置阶段就被拦住，不会等到发布产品时才报一条看不懂的 SQL 错误。
同一个物模型内部，标识还必须全局唯一——属性和功能、事件、标签之间也不能重名：
// 检查标识Key是否存在，物模型模块下唯一
func checkExistKey(key string, tsl model.TSL) bool {
for _, v := range tsl.Properties {
if strings.EqualFold(v.Key, key) {
return true
}
}
for _, v := range tsl.Functions {
if strings.EqualFold(v.Key, key) {
return true
}
}
// events、tags 同理
return false
}
strings.EqualFold 而不是 ==，说明大小写不敏感也被算进去了。对使用者来说这是件好事：Temperature 和 temperature 在产品里只能存在一个，不会出现两个语义相同、列名不同的属性。
---
二、一条 CREATE STABLE 里的三次约定
产品的物模型配完、点发布，平台会去时序库里建一张超级表。这段代码是整个设计的核心：
// pkg/tsd/internal/tdengine/tdengine.go
func (m *TdEngine) CreateDeviceStable(tsl *model.TSL) error {
// 属性字段
columns := []string{"ts TIMESTAMP"}
for _, v := range tsl.Properties {
maxLength := 0
if v.ValueType.TSLParamBase.MaxLength != nil {
maxLength = *v.ValueType.TSLParamBase.MaxLength
}
columns = append(columns, m.column(v.ValueType.Type, v.Key, maxLength))
// 属性上报时间
columns = append(columns, m.column("date", v.Key+"_time", 0))
}
// 标签字段
tags := make([]string, len(tsl.Tags)+1)
tags[0] = "device VARCHAR(255)"
for i, v := range tsl.Tags {
maxLength := 0
if v.ValueType.TSLParamBase.MaxLength != nil {
maxLength = *v.ValueType.TSLParamBase.MaxLength
}
tags[i+1] = m.column(v.ValueType.Type, v.Key, maxLength, 1)
}
tConent := ""
if len(tags) > 0 {
tConent = fmt.Sprintf("TAGS (%s)", strings.Join(tags, ","))
}
table := comm.ProductTableName(tsl.Key)
sqlStr := fmt.Sprintf("CREATE STABLE IF NOT EXISTS %s.%s (%s) %s", m.Option.Database, table, strings.Join(columns, ","), tConent)
_, err := m.exec(sqlStr)
return err
}
短短三十行里有三次约定，每一次都对着一个具体的现场问题。
第一次约定：一个属性两列。 p_temperature 存值，p_temperature_time 存这个值发生的时间。为什么不能只靠 ts 主时间列？因为设备是汇聚上报的，一批数据可能是补传、可能是缓存重发，同一批里的每个值未必发生在同一毫秒。把「值」和「值的时间」绑成一对，才敢在主时间列上按入库顺序写，同时保住每个测点自己的时刻。
第二次约定：类型跟着物模型走。
func (m *TdEngine) column(fType, key string, maxLength int, isTag ...int) string {
tdType := ""
switch fType {
case "int":
tdType = "INT"
case "long":
tdType = "BIGINT"
case "float":
tdType = "FLOAT"
case "double":
tdType = "DOUBLE"
case "string":
if maxLength == 0 {
maxLength = 150
}
tdType = "NCHAR(" + strconv.Itoa(maxLength) + ")"
case "boolean":
tdType = "BOOL"
case "date":
tdType = "TIMESTAMP"
default:
if maxLength == 0 {
maxLength = 150
}
tdType = "NCHAR(" + strconv.Itoa(maxLength) + ")"
}
...
}
字符串没有指定长度时默认 NCHAR(150)，而不是等到插入时才发现超长；long 走 BIGINT，不会因为设备报了个 13 位时间戳就被截断。类型这件事在建模阶段定一次，之后所有设备共用，写入侧不需要做任何类型判断。
第三次约定：命名空间前缀。 表名带 product_、device_、log_，属性列带 p_，标签列带 t_，全部转小写、中划线转下划线：
// pkg/tsd/comm/consts.go
const (
TdProductPrefix = "product_"
TdDevicePrefix = "device_"
TdLogPrefix = "log_"
TdPropertyPrefix = "p_"
TdTagPrefix = "t_"
)
func TsdColumnName(key string) string {
key = strings.ToLower(key)
if key == "ts" {
return "ts"
}
return TdPropertyPrefix + key
}
前缀不是为了好看。用户可能把属性叫 ts、叫 device、叫 status，这些词在时序库里都有特殊含义或已被平台占用。加一层前缀，属性命名空间和系统列就彻底分开了，ts 保留给主时间列、device 保留给设备标识 TAG，其余一律加前缀。
这三条约定合起来，超级表就变成一个稳定的契约：同一个产品的所有设备，写同一张表、同一组列，聚合查询天然对齐。
一台设备接入时，平台再基于这张超级表建子表：
// CreateDeviceTable 创建设备子表
func (m *TdEngine) CreateDeviceTable(table, stable string) error {
tag := table
stable = comm.ProductTableName(stable)
table = comm.DeviceTableName(table)
sqlStr := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING %s (device) TAGS ('%s')", table, stable, tag)
_, err := m.exec(sqlStr)
return err
}
产品是超级表、设备是子表，正好对上 TDengine 的一设备一表模型。查询时按设备过滤是子表扫描，按产品做趋势聚合是超级表扫描，两种访问模式都不需要额外索引。
---
三、改物模型，就是改表结构
物模型能改，但改的时候表结构得跟着动，否则新属性写不进去。SagooIoT 的做法是：元数据和存储放在同一个事务里改。
// internal/logic/product/dev_tsl_property.go AddProperty 片段
tsl.Properties = append(tsl.Properties, in.TSLProperty)
metaData, _ := json.Marshal(tsl)
err = dao.DevProduct.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
_, err = dao.DevProduct.Ctx(ctx).
Data(dao.DevProduct.Columns().Metadata, string(metaData)).
Where(dao.DevProduct.Columns().Key, in.ProductKey).
Update()
if err != nil {
return err
}
if tsdEnable {
tsdDb := tsd.DB()
if tsdDb == nil {
return errors.New("时序数据库连接失败")
}
if p.MetadataTable == 1 {
maxLength := 0
if in.ValueType.TSLParamBase.MaxLength != nil {
maxLength = *in.ValueType.TSLParamBase.MaxLength
}
err = tsdDb.AddField(p.Key, in.Key, in.ValueType.Type, maxLength)
if err != nil {
return err
}
}
}
return nil
})
AddField 里同样是一对操作，加值列之后立刻加时间列：
func (m *TdEngine) AddField(table, field string, fType string, len int) error {
table = comm.ProductTableName(table)
sqlStr := fmt.Sprintf("ALTER STABLE %s.%s ADD COLUMN %s", m.Option.Database, table, m.column(fType, field, len))
_, err := m.exec(sqlStr)
if err == nil {
sqlStr = fmt.Sprintf("ALTER STABLE %s.%s ADD COLUMN %s", m.Option.Database, table, m.column("date", field+"_time", 0))
_, err = m.exec(sqlStr)
}
return err
}
两件事一起成功、一起回滚，就不会出现「物模型里有这个属性、表里没这列」的中间态。批量的 BatchAddProperty 也是同一个事务里循环加列，导入一份 TSL 文件时不会留下半张表。
改也不是无限制的。三个写操作开头都有一句相同的判断：
if p.Status != 0 {
return errors.New("产品已发布,无法新增") // 或"无法修改"、"无法删除"
}
产品发布之后物模型就锁定了。这个取舍很符合工业现场：表结构变更会影响历史数据的可比性，一条趋势曲线如果前半段来自 FLOAT、后半段来自 NCHAR，看图的人会莫名其妙。所以改结构必须发生在设备大规模接入之前；已经上线的产品要演进，就走新建产品、迁移设备这条路，历史数据各归各表。
删除属性是这几个操作里最重的。如果删的是最后一个属性，超级表就没有业务列了，平台会把整张表重建：
if plen > 1 {
if err = tsdDb.DelField(p.Key, in.Key); err != nil {
return err
}
} else {
// 删除超级表
if err = tsdDb.DropDeviceStable(p.Key); err != nil {
return err
}
// 删除子表，并把设备置为未启用
devList, err := service.DevDevice().GetAllForProduct(ctx, p.Key)
...
existTable = 0
existStatus = model.ProductStatusOff
}
不是留一张空表在那里等人踩，而是把产品状态、设备状态、metadata_table 标志一起复位，回到「没建过表」的干净状态。配置文件和数据库状态始终一致，不会出现悬空的表。
---
四、报文进来先认 method
表结构定好了，接下来是数据怎么进去。设备发上来的是一段 JSON，平台第一步不是取值，而是认这个方法名：
// internal/logic/product/dev_tsl_parse.go
func (s *sDevTSLParse) ParseData(ctx context.Context, device *model.DeviceOutput, data []byte) (iotModel.ReportPropertyData, error) {
if len(data) == 0 {
return nil, errors.New("data is empty")
}
if device == nil {
return nil, errors.New("device not found")
}
var reportData sagooProtocol.ReportPropertyReq
if err := json.Unmarshal(data, &reportData); err != nil {
g.Log().Debugf(ctx, "parse data error: %s, message:%s, message ignored", err, string(data))
return nil, err
}
if len(reportData.Params) == 0 {
return nil, errors.New("report data is empty")
}
switch reportData.Method {
case "thing.event.property.post":
return s.HandleProperties(ctx, device, reportData.Params)
case "thing.event.property.pack.post":
return s.handlePackPost(ctx, device, reportData.Params)
default:
return nil, errors.New("unsupported method")
}
}
thing.event.property.post 是单设备直报，params 里就是属性键值。thing.event.property.pack.post 是打包上报，一个网关把自己和挂载的子设备的数据打在一个包里发上来：
case "properties":
properties, ok := value.(map[string]interface{})
if !ok {
g.Log().Debug(ctx, "Invalid properties data format")
continue
}
if prop, err := s.HandleProperties(ctx, device, properties); err == nil {
allProperties = mergePropertyData(allProperties, prop)
}
case "subDevices":
subDevices, ok := value.([]interface{})
...
for _, sub := range subDevices {
handleSubDeviceProperties(ctx, device, subData, &allProperties)
}
子设备那段里有个身份校验，identity.deviceKey 必须和当前连接设备一致才会解析：
deviceKey, ok := identity["deviceKey"].(string)
if !ok || deviceKey != device.Key {
g.Log().Debug(ctx, "Invalid or unmatched device key")
return
}
也就是说，网关不能替别人上报数据，报文里的身份必须和 TCP/MQTT 连接上认证过的身份对得上。设备身份由连接决定，报文只能声明自己的数据，不能声明自己是谁——这条边界放在解析入口，比放在业务逻辑里可靠得多。
---
五、十一个转换器：让脏数据进不了库
真正把 "23.456789" 变成 23.46 的，是这一组类型转换器。入口只有一个：
func (t TSLValueType) ConvertValue(v interface{}) interface{} {
var transfer Transfer
switch t.Type {
case consts.TypeInt:
transfer = TInt(t.TSLParam)
case consts.TypeLong:
transfer = TLong(t.TSLParam)
case consts.TypeFloat:
transfer = TFloat(t.TSLParam)
...
case consts.TypeArray:
transfer = TArray(t.TSLParam)
case consts.TypeObject:
transfer = TObject(t.TSLParam)
default:
return nil
}
return transfer.Convert(v)
}
type Transfer interface {
Convert(interface{}) interface{}
}
十一种类型各实现一个 Convert，每种都在做同一件事：把设备给的任意值，收敛成物模型承诺的那个形状。
| 类型 | 收敛规则 | 代码要点 |
|---|---|---|
| int | 转整数后夹紧到 [min, max] | 越界返回边界值 |
| long | 转 int64 后夹紧 | 同上 |
| float | 夹紧 → 按 decimals 舍入 → 降为 float32 | 默认两位小数 |
| double | 夹紧 → 按 decimals 舍入 | 保留 float64 |
| string | 转字符串 → 超 maxLength 截断 | 按字节长度 |
| boolean | 转 bool → 映射 trueValue/falseValue | 支持反转语义 |
| date | 按两种 layout 自适应解析 | 秒级/毫秒级 |
| timestamp | 毫秒时间戳 → 格式化字符串 | UnixMilli |
| enum | 命中枚举值才返回，否则空串 | 枚举白名单 |
| array | 逐元素递归 ConvertValue | 元素类型可声明 |
| object | 按 Properties 逐字段递归 | 结构体嵌套 |
两个转换器值得展开看。浮点这边：
type TFloat TSLParam
func (tFloat TFloat) Convert(v interface{}) interface{} {
number := gconv.Float64(v)
if tFloat.TSLParamBase.Min != nil && float32(*tFloat.TSLParamBase.Min) > float32(number) {
number = float64(*tFloat.TSLParamBase.Min)
}
if tFloat.TSLParamBase.Max != nil && float32(*tFloat.TSLParamBase.Max) < float32(number) {
number = float64(*tFloat.TSLParamBase.Max)
}
defaultDecimal := 2
if tFloat.TSLParamBase.Decimals != nil {
defaultDecimal = *tFloat.TSLParamBase.Decimals
}
number32, _ := strconv.ParseFloat(strconv.FormatFloat(number, 'f', defaultDecimal, 64), 32)
return float32(number32)
}
一个温度传感器接反了、报了个 -273.15 以下的数，或者模块故障吐出 9999，转换器会把它夹到量程边界上。这样库里不会出现离谱的尖峰把整条趋势图的纵轴拉爆，看图的人也不必每次都先怀疑传感器。小数位则是先按 decimals 格式化、再降为 float32，保证「库里存的位数」和「界面上看到的位数」一致。
枚举这边更直接：
func (tEnum TEnum) Convert(v interface{}) interface{} {
tE := gconv.String(v)
if tEnum.TSLParamExtension.Elements == nil {
return ""
}
for _, node := range tEnum.TSLParamExtension.Elements {
if node.Value == tE {
return node.Value
}
}
return ""
}
设备报了个协议之外的枚举值（固件升级多了一个状态码），这个值不会进库，库里只会留下空值。空值在统计上是可识别的缺失，比塞进一个没人认识的字符串好处理得多。
布尔类型还有个实用设计——真值和假值可以配置：
type TBoolean TSLParam
func (tBoolean TBoolean) Convert(v interface{}) interface{} {
b := gconv.Bool(v)
if tBoolean.TSLParamBase.TrueValue != nil && *tBoolean.TSLParamBase.TrueValue == b {
return true
}
if tBoolean.TSLParamBase.FalseValue != nil && *tBoolean.TSLParamBase.FalseValue == b {
return false
}
return b
}
有的设备用 1/0，有的用 "ON"/"OFF"，有的高电平代表断开。物模型里把语义写在 trueText/falseText 上，展示层用它渲染「是/否」「开/关」，库里统一是 BOOL。协议差异在边缘收敛，数据在中台统一，这是这一层存在的意义。
---
六、一个属性两个时间是怎么填的
前面说每个属性两列，那 _time 列的值从哪来？看这个函数：
// CreatePropertyNode 创建属性节点
func (s *sDevTSLParse) CreatePropertyNode(property *model.TSLProperty, value interface{}, defaultTime time.Time) iotModel.ReportPropertyNode {
var timestamp int64
var val interface{}
if mapInfo, ok := value.(map[string]interface{}); ok {
if timeVal, timeOK := mapInfo["time"].(float64); timeOK {
timestamp = int64(timeVal)
} else {
timestamp = defaultTime.Unix()
}
val = property.ValueType.ConvertValue(mapInfo["value"])
} else {
timestamp = defaultTime.Unix()
val = property.ValueType.ConvertValue(value)
}
return iotModel.ReportPropertyNode{
CreateTime: timestamp,
Value: val,
}
}
设备可以报两种形态：裸值 "temperature": 23.5，或者带时间的对象 "temperature": {"time": 1735689600, "value": 23.5}。前者用服务器接收时刻补齐，后者用设备自己给的时间戳。
两种写法都支持，落到库里就是「主时间列 = 入库时刻」和「p_temperature_time = 设备声称的时刻」两套。这个设计在排查现场问题时特别有用：设备时钟漂了、数据补传了、边缘侧缓存积压了几小时才发上来，两条时间线一比就清楚——数据是什么时候到的，和它自己认为发生在什么时候，是两件事。
---
七、同一份报文，为什么要解析两次
在接入链路上，一条属性上报进来后，处理函数大致是这样排的：
// network/core/logic/model/up/property/reporter/reporter.go
go baseLogic.InertTdLog(ctx, consts.MsgTypePropertyReport, data.DeviceDetail.Key, reportData.Id, reportData)
go baseLogic.UpdateShadow(ctx, consts.MsgTypePropertyReport, data.ProductKey, data.DeviceDetail, reportData.Params)
//解析数据
reportDataInfo, err := service.DevTSLParse().ParseData(ctx, data.DeviceDetail, data.PayLoad)
if err != nil {
return err
}
//北向属性上报消息
north.WriteMessage(ctx, north.PropertyReportMessageTopic, nil, data.ProductKey, data.DeviceDetail.Key, iotModel.PropertyReportMessage{
Properties: reportDataInfo,
})
//告警处理
err = service.AlarmRule().Check(ctx, data.ProductKey, data.DeviceKey, consts.AlarmTriggerTypeProperty, reportDataInfo)
// 场景规则投递到队列
这里已经解析过一次了，得到 reportDataInfo 给告警、北向、场景规则用。而那行 InertTdLog 只是把原始报文按日志写走，不解析。半小时后，落库任务从队列里取出一批日志，又解析了一次：
// internal/queues/deviceDataTsdSave.go
for _, devLog := range items {
if devLog.Type == consts.MsgTypePropertyReport {
device, err := deviceCache.GetDeviceDetailInfo(devLog.Device)
if err != nil {
continue
}
deviceData, err := service.DevTSLParse().ParseData(context.Background(), device, []byte(devLog.Content))
if err != nil {
continue
}
deviceDataList[devLog.Device] = append(deviceDataList[devLog.Device], deviceData)
}
}
if len(deviceDataList) > 0 {
db := tsd.GetDB()
defer db.Close()
_, err = db.BatchInsertMultiDeviceData(deviceDataList)
}
同一个函数被调用两次，不是重复劳动，而是两条路径各自的职责不同：
- 接入侧那次是实时路径。它要在毫秒级给出告警判定、要立刻响应设备 ACK、要把消息推给北向系统和场景规则。它不能被数据库拖住，所以算完就走。
- 落库侧那次是批量路径。它要攒批、要按设备分组、要一次 INSERT 多行。它拿的是日志原文，重新解析一遍，才能把同一设备的多条数据拼成一次写库。
好处是两条路径彻底解耦：数据库抖动或者批量任务积压，接入侧照常收数、照常告警，ACK 照常返回；反过来，接入侧重启也不影响队列里已经排着的落库任务。而 ParseData 本身是纯函数——输入是「设备详情 + 报文字节」，输出是「属性键值对」，同一个输入永远同一个输出，所以两次解析的结果一定一致，不存在实时侧和落库侧对不上的情况。
顺带一句，落库任务里用了 sync.Pool 复用 DeviceLog 对象，每批几百条日志不必反复分配：
deviceLogPool = sync.Pool{
New: func() interface{} {
return new(iotModel.DeviceLog)
},
}
---
八、写进去的时候是一条 INSERT 多行
批量路径最终落到 BatchInsertMultiDeviceData，SQL 是按设备拼的：
// pkg/tsd/internal/tdengine/device.go
table := comm.DeviceTableName(deviceKey)
keys, fieldList := comm.GetDeviceField(deviceDataList[0])
var (
ts = time.Now().UnixMilli()
baseSQL = "INSERT INTO " + table + " (" + strings.Join(fieldList, ",") + ") VALUES"
sqlBuilder strings.Builder
)
sqlBuilder.WriteString(baseSQL)
for i, row := range deviceDataList {
ts++
value := comm.GetDeviceValue(keys, row)
sqlBuilder.WriteString(fmt.Sprintf(" (%s)", "'"+time.UnixMilli(ts).Format(time.RFC3339Nano)+"',"+strings.Join(value, ",")))
if sqlBuilder.Len() > 15*1024 || i == len(deviceDataList)-1 {
_, err := m.db.Exec(strings.TrimRight(sqlBuilder.String(), " "))
sqlBuilder.Reset()
sqlBuilder.WriteString(baseSQL)
}
}
几个细节值得注意。
主时间列按毫秒递增。 ts++ 保证同一批数据在主时间列上严格有序，不会因为同一毫秒内多条数据导致覆盖或乱序。真实时刻在 _time 列里，主时间列只负责排序和分区。
SQL 按 15KB 分段。 不是一条数据一条 SQL，也不是无限拼接到内存爆掉。攒到 15KB 就发一次，既拿到了多行插入的吞吐，又把单条语句的大小控制住。
字段顺序取自第一条数据。 GetDeviceField 用第一行数据的 key 生成字段列表，GetDeviceValue 再按这个固定顺序取值：
func GetDeviceValue(field []string, data iotModel.ReportPropertyData) []string {
var value []string
//跟据统一的key列表顺序，对数据值排序输出
for _, key := range field {
for k, v := range data {
_, ok := data[key]
if ok {
if k == key {
value = append(value, "'"+gvar.New(v.Value).String()+"'")
value = append(value, "'"+gtime.New(v.CreateTime).Format("Y-m-d H:i:s")+"'")
}
}
}
}
return value
}
Go 的 map 遍历顺序是随机的，如果不先固定一份 key 序列，同一批数据的字段顺序就会飘，拼出来的 SQL 列和值容易错位。先取一份顺序、再按顺序取值，这个小动作是批量写入正确性的前提。
---
九、启动自检：表结构和物模型对不齐时自己补
再稳的流程也会有例外：换了时序库、迁移了环境、有人手工删了表。所以 dev_init 里有一段开机自检，带 Redis 锁避免多实例同时建：
var list []*entity.DevProduct
c := dao.DevProduct.Columns()
err = dao.DevProduct.Ctx(ctx).Where(c.Status, model.ProductStatusOn).Where(c.MetadataTable, 1).Scan(&list)
if err != nil || len(list) == 0 {
return
}
// 检测td表结构是否存在，不存在则创建
for _, p := range list {
stable := comm.ProductTableName(p.Key)
b, _ := tsdDb.CheckDeviceStable(stable)
if b {
continue
}
var tsl *model.TSL
err = json.Unmarshal([]byte(p.Metadata), &tsl)
if len(tsl.Properties) == 0 {
g.Log().Errorf(ctx, "产品 %s 物模型数据异常", p.Key)
continue
}
err = tsdDb.CreateDeviceStable(tsl)
}
物模型 JSON 就存在产品表的 metadata 字段里，是唯一事实来源。只要这份 JSON 还在，超级表随时可以重建——表结构是可推导的产物，不是需要备份的资产。这个性质让环境迁移变得很轻：数据库带过去、时序库空着启动就行，平台自己会把表建齐。
同样的自检也覆盖设备子表：metadata_table 标志是 1 但子表不见了，就补建一张。
---
十、一份数据结构，五个下游
回头看 reportDataInfo 这个 map[string]ReportPropertyNode，它被五处消费：
| 消费方 | 拿它做什么 | 对数据的期待 |
|---|---|---|
| 告警规则 | 阈值、越界、离线判定 | 已按量程夹紧、类型正确 |
| 设备影子 | 更新 reported 态 | 值与时间成对 |
| 时序库 | 批量落库 | 字段名与列名一一对应 |
| 场景规则 | 触发联动 | 结构稳定、可序列化 |
| 北向接口 | 转发给上层系统 | 与物模型定义一致 |
五处用的是同一个结构，意味着它们看到的数据一定是一样的。告警判的是 23.46，库里存的也是 23.46，大屏上显示的也是 23.46，不会出现「告警说超了、曲线看着没超」这种各说各话的场面。
这也是物模型这层抽象真正的价值：它把「这台设备报的字段叫什么、是什么类型、量程多少、单位是什么」从代码里挪到了配置里。接一个新厂家的设备，不用改解析代码，只要在物模型里把属性建好，再写一段脚本把厂家的私有报文转成物模型的键——脚本那层在另一篇文章里讲过，这里只说结果：同一个产品的设备，无论来自哪个厂家、走哪种协议，落到库里都是同一张表、同一组列。
---
十一、这套设计的收益
把上面的链路串起来看，几个收益是明确的。
加属性不用动代码。 产品详情里加一行属性，事务里自动 ALTER STABLE ADD COLUMN 两列，从下一条上报开始就能存。设备端不需要升级协议、不需要改字段顺序、不需要重启。
换厂家不用改表。 报文差异被脚本层吸收，物模型是稳定的中间层。同一批传感器换了供应商，历史曲线照样能连起来看。
数据可比。 同一产品的所有设备共用超级表和列定义，跨设备对比、按产品聚合、按标签分组，SQL 都是一句话的事。
脏数据在入口就被收敛。 类型转换、量程夹紧、枚举白名单、字符串截断，四道处理全部发生在解析阶段。库里的类型永远是物模型承诺的类型，下游不用做防御性判断。
边界清晰。 产品发布后物模型锁定，结构变更与数据写入彻底分开：变更只在配置阶段发生一次，写入路径上没有任何 DDL。这也是写入侧能放心攒批、拼 15KB 大 SQL 的前提。
如果要给这套设计下一个定义，它更像用一份 schema 驱动三层：向上驱动界面表单和报表，向中驱动报文解析和校验，向下驱动时序库的表结构与列类型。三层共用一份定义，才有资格叫「物模型」，否则只是几张配置表。
---
结语
物联网平台里最容易出问题的地方，往往不是协议对接，而是「同一个物理量在不同环节有不同解释」：设备说 235（0.1℃ 为单位），平台说 23.5℃，数据库存 23（INT），大屏显示 24。每一层都做了自己的四舍五入，最后一圈下来谁也对不上。
SagooIoT 的处理方式是把解释权收归一处——物模型说它是 float、两位小数、量程 -40~85，那么解析、告警、影子、时序库、北向全按这个口径来。表结构是它算出来的，值是它裁剪过的，时间是它补齐的。
一份配置，管住了从报文到曲线的整条路。
---
项目地址
- 文档站：https://iotdoc.sagoo.cn
- 社区版源码：https://github.com/orgs/sagoo-cloud/repositories
- 本文涉及的实现取自本地专业版发行版源码：
internal/logic/product（物模型增删改与解析）、internal/model/dev_tsl*.go（TSL 结构与类型转换器）、pkg/tsd（TDengine 建表、加列、批量写入）、internal/queues/deviceDataTsdSave.go（落库队列）、network/core/logic/model/up/property（接入侧上行处理）。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/microrain/article/details/169520139
