破解物联网平台性能瓶颈：TDengine 在联犀的应用与优化

TDengine 2024-12-05

小T导读

本篇文章是“2024，我想和 TDengine 谈谈”征文活动的三等奖作品。文章从一个具体的业务场景出发，分析了企业在面对海量时序数据时的挑战，并提出了利用 TDengine 高效处理和存储数据的方法，帮助企业解决在数据采集、存储、分析等方面的痛点。通过这篇文章，作者不仅展示了自己对数据处理技术的理解，还进一步阐释了时序数据库在行业中的潜力与应用价值，为读者提供了很多实际的操作思路和技术选型的参考。

在当前的物联网行业，TDengine 已成为国内占有率最高的开源时序数据库，并在开源物联网平台中扮演着至关重要的角色。接下来，我们将分几个模块详细探讨联犀物联网平台如何与 TDengine 深度结合，并进一步提升 TDengine 的扩展性，推动其在物联网场景中的应用潜力。

物模型如何结合 TDengine？

物模型介绍

现实世界由众多真实存在的物理设备组成，我们可以将这些设备称之为“物”。物联网的目标是通过网络将这些“物”连接在一起，并将其数字化为云端服务或资源，从而实现智能化应用。因此，在物联网构建的数字世界中，首先需要对“物”进行清晰、统一的定义，明确其功能和能提供的服务资源。

ICA 联盟从产品层面对“物”进行了功能建模，提出了统一的“物的抽象模型”和“物的描述语言”（TSL，Things Specification Language）。物的抽象模型描述了“设备是什么”以及“设备能做什么”，包括物的状态、档案信息和功能定义等方面。

在物模型中，最为关键的是属性（property）。以智能电灯为例，其状态具有二元性：开启或关闭。用户可以通过控制操作轻松在这两种状态之间切换。此外，一些智能电灯还具备更多高级功能，允许用户根据个人需求调整亮度、颜色和色温等参数。

智能设备的属性通常具备读写能力，这意味着应用程序不仅可以读取设备的当前状态，还可以修改属性来调整设备行为。例如，在环境监测设备中，应用程序可以读取温度和湿度数据，并根据需要调整参数，以适应不同的环境条件。

物模型如何结合 TDengine？

物模型超级表与产品的关系

在联犀物联网平台中，存在两种物模型：公共物模型和产品物模型。公共物模型是由多个产品共同定义的共享物模型，而产品物模型则为每个特定产品设备单独定义。定义物模型时，通常会有以下两种操作：

产品物模型：每类特定产品设备使用一个超级表来定义。
通用物模型：只创建一个超级表，后续的产品可以按需引入该物模型，产品下的设备则直接使用这个超级表。

TDengine 支持灵活的数据模型设计，既可以使用多列模型，也可以选择单列模型。多列模型相当于将多个字段存储在同一张超级表中，通常在写入和存储效率上表现较优。而在某些情况下，例如数据采集点的种类和数量经常变化时，单列模型则可能更为适用，因为它简化了应用程序的设计和管理，允许独立管理和扩展每个物理量的超级表。

综上所述，TDengine 提供了灵活的数据模型选项，用户可以根据具体需求和应用场景选择最适合的模型。无论是采用窄表设计、单列模型还是多列模型，最终目的是为了优化性能和简化管理复杂性。

物模型和表分为三种对应关系

普通类型： 根据物模型的数据类型，映射单列模式的超级表。物模型定义示例如下图所示。

示例 SQL 如下：

CREATE STABLE IF NOT EXISTS model_custom_property_00b_int (ts timestamp,param BIGINT) 
  TAGS (product_id BINARY(50),device_name BINARY(50),property_type BINARY(50));

相关参数说明：

字段	类型	说明
model_custom_property_00b_switchg	超级表名	组成为模型_物模型的类型(自定义物模型,通用物模型)_属性_产品ID_属性名称
ts	表字段	时间戳,主键
param	参数	数组每一位的数据都存于此
_num	数组的下角标	表示数据属于哪一位
product_id	产品id	设备所属的产品ID
device_name	设备名	同时也叫sn
property_type	属性的类型	在获取数据的时候方便序列化数据

可以看到，数组类型的定义与简单类型相似，区别在于它额外添加了一个 _num
字段，用来标识数组的下标。这种设计的好处在于，即使数组长度达到 1000，依然只需定义一个超级表。虽然每个设备的普通表需要定义 1000 个，但由于是由一个超级表进行管理，整体管理变得更加简便。接下来，我们来看看每个设备创建的普通表是什么样的。

CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_0 USING model_custom_property_00b_switchg  TAGS('00b','device1',0,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_1 USING model_custom_property_00b_switchg  TAGS('00b','device1',1,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_2 USING model_custom_property_00b_switchg  TAGS('00b','device1',2,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_3 USING model_custom_property_00b_switchg  TAGS('00b','device1',3,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_4 USING model_custom_property_00b_switchg  TAGS('00b','device1',4,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_5 USING model_custom_property_00b_switchg  TAGS('00b','device1',5,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_6 USING model_custom_property_00b_switchg  TAGS('00b','device1',6,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_7 USING model_custom_property_00b_switchg  TAGS('00b','device1',7,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_8 USING model_custom_property_00b_switchg  TAGS('00b','device1',8,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_9 USING model_custom_property_00b_switchg  TAGS('00b','device1',9,'bool');

如何发挥 TDengine 的性能？

在设备上报信息时，如果在线设备达到 120 万，每个设备每 10 分钟上报一次数据，那么每秒钟大约会有 2000 条消息需要处理。每条消息不仅需要入库一条调试日志，还需要记录一条属性历史数据。通过大量性能测试和优化，我们发现，突破 2000 QPS 是非常具有挑战性的。深入分析了 TDengine 的体系架构后，我们实施了以下优化措施：

从 HTTP 切换到 WebSocket：TDengine 从 3.x 版本开始支持 WebSocket，我们顺势升级，经过测试，性能和稳定性都有所提升，但在高并发的情况下，系统资源消耗依然较大。
改进同步操作为异步操作：通过将同步操作转为异步处理，显著提高了整体性能。
SQL 批量插入优化：TDengine 支持单条 SQL 语句插入多条数据，甚至跨表插入，这大大提升了写入性能。

官方语法如下：

INSERT INTO
    tb_name
        [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
        [(field1_name, ...)]
        VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
    [tb2_name
        [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
        [(field1_name, ...)]
        VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
    ...];


INSERT INTO tb_name [(field1_name, ...)] subquery

官方示例如下：
INSERT INTO d21001 USING meters TAGS ('California.SanFrancisco', 2) VALUES ('2021-07-13 14:06:34.630', 10.2, 219, 0.32) ('2021-07-13 14:06:35.779', 10.15, 217, 0.33)
            d21002 USING meters (groupId) TAGS (2) VALUES ('2021-07-13 14:06:34.255', 10.15, 217, 0.33)
            d21003 USING meters (groupId) TAGS (2) (ts, current, phase) VALUES ('2021-07-13 14:06:34.255', 10.27, 0.31);

联犀利用这一语法实现了异步操作，操作流程如下：

设备插入数据时，首先生成类似 d21003 USING meters (groupId) TAGS (2) (ts, current, phase) VALUES ('2021-07-13 14:06:34.255', 10.27, 0.31)
格式的插入语句。
将生成的 SQL 语句放入 Golang 的 Channel 中。
多个异步入库协程监听 Channel，并从中取出 SQL 语句。当执行间隔超过 1 秒或 SQL 数量达到上限时，协程会将这些数据合并成完整的 SQL 批量插入数据库。

由于设备上报数据的整个流程不涉及磁盘操作，整个过程速度非常快。即使在未完全优化的情况下，经过测试，单机配置为 8 核 16GB 的服务器，能够稳定支持每秒 7000 并发请求，且保持毫秒级的低延迟，并无任何错误。

关键代码如下：



type Td struct {
    *sql.DB
}


type ExecArgs struct {
    Query string
    Args  []any
}


var (
    td         = Td{}
    once       = sync.Once{}
    insertChan = make(chan ExecArgs, 1000)
)


const (
    asyncExecMax = 200 //异步执行sql最大数量
    asyncRunMax  = 40
)


func NewTDengine(DataSource conf.TSDB) (TD *Td, err error) {
    once.Do(func() {
        if DataSource.Driver == "" {
            DataSource.Driver = "taosWS"
        }
        td.DB, err = sql.Open(DataSource.Driver, DataSource.DSN)
        if err != nil {
            return
        }
        td.DB.SetMaxIdleConns(50)
        td.DB.SetMaxOpenConns(50)
        td.DB.SetConnMaxIdleTime(time.Hour)
        td.DB.SetConnMaxLifetime(time.Hour)
        _, err = td.Exec("create database if not exists ithings;")
        if err != nil {
            return
        }
        for i := 0; i < asyncRunMax; i++ {
            utils.Go(context.Background(), func() {
                td.asyncInsertRuntime()
            })
        }
    })
    if err != nil {
        logx.Errorf("TDengine 初始化失败,err:%v", err)
    }
    return &td, err
}


func (t *Td) asyncInsertRuntime() {
    r := rand.Intn(1000)
    tick := time.Tick(time.Second/2 + time.Millisecond*time.Duration(r))
    execCache := make([]ExecArgs, 0, asyncExecMax*2)
    exec := func() {
        if len(execCache) == 0 {
            return
        }
        sql, args := t.genInsertSql(execCache...)
        var err error
        for i := 3; i > 0; i-- { //三次重试
            _, err = t.Exec(sql, args...)
            if err == nil {
                break
            }
        }
        if err != nil {
            logx.Error(err)
        }
        execCache = execCache[0:0] //清空切片
    }
    for {
        select {
        case _ = <-tick:
            exec()
        case e := <-insertChan:
            execCache = append(execCache, e)
            if len(execCache) > asyncExecMax {
                exec()
            }
        }
    }


}


func (t *Td) AsyncInsert(query string, args ...any) {
    insertChan <- ExecArgs{
        Query: query,
        Args:  args,
    }
}
func (t *Td) genInsertSql(eas ...ExecArgs) (query string, args []any) {
    qs := make([]string, 0, len(eas))
    as := make([]any, 0, len(eas))
    for _, e := range eas {
        qs = append(qs, e.Query)
        as = append(as, e.Args...)
    }
    return fmt.Sprintf("insert into %s;", strings.Join(qs, " ")), as
}

TDengine 查询

数据插入数据库后，我们开始着手进行数据查询和展示。得益于 TDengine 支持多种丰富的数据聚合方式，数据分析变得更加高效和便捷。

ORM 设计

灵活的查询方式离不开 ORM 框架的支持。然而，TDengine 的官方库并未提供 ORM 框架，开源社区也没有专门为 TDengine 开发的 ORM 框架。为了填补这一空白，联犀基于知名 ORM 框架 Squirrel（https://github.com/Masterminds/squirrel）进行了扩展，以支持 TDengine 的语法。以下是 ORM 的示例：


func (d *DeviceDataRepo) getPropertyArgFuncSelect(
    ctx context.Context,
    filter msgThing.FilterOpt) (sq.SelectBuilder, error) {
    schemaModel, err := d.getSchemaModel(ctx, filter.ProductID)
    if err != nil {
        return sq.SelectBuilder{}, err
    }
    p, ok := schemaModel.Property[filter.DataID]
    if !ok {
        return sq.SelectBuilder{}, errors.Parameter.AddMsgf("dataID:%s not find", filter.DataID)
    }
    var (
        sql sq.SelectBuilder
    )


    if p.Define.Type == schema.DataTypeStruct {
        sql = sq.Select("FIRST(ts) AS ts", d.GetSpecsColumnWithArgFunc(p.Define.Specs, filter.ArgFunc))
    } else {
        sql = sq.Select("FIRST(ts) AS ts", fmt.Sprintf("%s(param) as param", filter.ArgFunc))
    }
    if filter.Interval != 0 {
        sql = sql.Interval("?a", filter.Interval) //TDengine特有语法
    }
    if len(filter.Fill) > 0 {
        sql = sql.Fill(filter.Fill)//TDengine特有语法
    }
    return sql, nil
}

灵活的查询接口

借助上述的 ORM 底层实现，我们才得以实现灵活的查询接口，下面是查询接口示例。

请求参数:

回复参数：

TDengine 链路追踪

最后，我们还需要解决链路追踪的问题：我们需要记录每个 SQL 的执行耗时，并能够与业务链路 ID 打通，确保可以通过日志追溯到具体的 SQL 执行者和执行过程。虽然官方驱动较为底层，但它支持通过 context 进行传递。联犀的链路追踪同样基于 context 传递，因此我们只需在执行 SQL 时记录相关日志即可。无论是采用 HTTP、WebSocket 还是 CGo 连接方式，日志都统一记录在 driver-go/taosWS/connection.go
文件中。联犀打印的日志如下: