14 KiB

Raw Blame History

ThingsBoard 数据存储源码分析

1. 概述

ThingsBoard 的数据存储层采用分层设计，支持多种数据库存储不同类型的任务。实体数据存储在关系型数据库（PostgreSQL），时序数据可以存储在 PostgreSQL/TimescaleDB 或 Cassandra。

2. 数据存储架构

2.1 存储分层

应用层
  ├── TelemetryService (遥测服务)
  ├── AttributeService (属性服务)
  └── EntityService (实体服务)
         ↓
数据访问层 (DAO)
  ├── TimeseriesDao (时序数据)
  ├── AttributeDao (属性数据)
  └── EntityDao (实体数据)
         ↓
数据库层
  ├── PostgreSQL (实体数据 + 时序数据)
  ├── TimescaleDB (时序数据扩展)
  └── Cassandra (时序数据)

2.2 数据分类

实体数据: 租户、用户、设备、资产等
- 存储位置: PostgreSQL
- 特点: 结构化数据，需要事务支持
时序数据: 设备遥测数据
- 存储位置: PostgreSQL/TimescaleDB 或 Cassandra
- 特点: 时间序列数据，高写入量
属性数据: 设备属性（服务器端、客户端、共享）
- 存储位置: PostgreSQL
- 特点: 键值对数据
最新值数据: 时序数据的最新值
- 存储位置: PostgreSQL
- 特点: 快速查询最新值

3. 时序数据存储

3.1 时序数据服务

位置: application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java

关键方法:

/**
 * 保存时序数据
 */
@Override
public void saveTimeseries(TimeseriesSaveRequest request) {
    TenantId tenantId = request.getTenantId();
    EntityId entityId = request.getEntityId();
    
    // 检查是否为内部实体
    checkInternalEntity(entityId);
    
    boolean sysTenant = TenantId.SYS_TENANT_ID.equals(tenantId) || tenantId == null;
    
    // 检查是否启用数据库存储
    if (sysTenant || !request.getStrategy().saveTimeseries() || 
        apiUsageStateService.getApiUsageState(tenantId).isDbStorageEnabled()) {
        
        // 验证数据
        KvUtils.validate(request.getEntries(), valueNoXssValidation);
        
        // 保存时序数据
        ListenableFuture<TimeseriesSaveResult> future = saveTimeseriesInternal(request);
        
        if (request.getStrategy().saveTimeseries()) {
            Futures.addCallback(future, getApiUsageCallback(tenantId, 
                request.getCustomerId(), sysTenant), tsCallBackExecutor);
        }
    } else {
        request.getCallback().onFailure(
            new RuntimeException("DB storage writes are disabled due to API limits!"));
    }
}

3.2 时序数据 DAO

位置: dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java

关键方法:

/**
 * 保存时序数据
 */
private ListenableFuture<TimeseriesSaveResult> doSave(
    TenantId tenantId, 
    EntityId entityId, 
    List<TsKvEntry> tsKvEntries, 
    long ttl, 
    boolean saveLatest, 
    boolean saveTs) {
    
    // 1. 准备分区保存任务
    List<ListenableFuture<Integer>> tsFutures = saveTs ? 
        new ArrayList<>(tsKvEntries.size() * INSERTS_PER_ENTRY_WITHOUT_LATEST) : null;
    
    // 2. 准备最新值保存任务
    List<ListenableFuture<Long>> latestFutures = saveLatest ? 
        new ArrayList<>(tsKvEntries.size()) : null;
    
    // 3. 为每个数据点创建保存任务
    for (TsKvEntry tsKvEntry : tsKvEntries) {
        if (saveTs) {
            // 保存分区信息
            tsFutures.add(timeseriesDao.savePartition(
                tenantId, entityId, tsKvEntry.getTs(), tsKvEntry.getKey()));
            
            // 保存时序数据
            tsFutures.add(timeseriesDao.save(tenantId, entityId, tsKvEntry, ttl));
        }
        
        if (saveLatest) {
            // 保存最新值
            latestFutures.add(Futures.transform(
                timeseriesLatestDao.saveLatest(tenantId, entityId, tsKvEntry), 
                version -> {
                    if (version != null) {
                        // 通知 EDQS 服务更新
                        TenantId edqsTenantId = entityId.getEntityType() == EntityType.TENANT ? 
                            (TenantId) entityId : tenantId;
                        edqsService.onUpdate(edqsTenantId, ObjectType.LATEST_TS_KV, 
                            new LatestTsKv(entityId, tsKvEntry, version));
                    }
                    return version;
                }, MoreExecutors.directExecutor()));
        }
    }
    
    // 4. 合并所有 Future
    ListenableFuture<Integer> dpsFuture = saveTs ? 
        Futures.transform(Futures.allAsList(tsFutures), SUM_ALL_INTEGERS, 
            MoreExecutors.directExecutor()) : Futures.immediateFuture(0);
    
    // 5. 返回结果
    return Futures.transform(
        Futures.allAsList(Arrays.asList(dpsFuture, versionsFuture)),
        results -> new TimeseriesSaveResult(...),
        MoreExecutors.directExecutor());
}

3.3 SQL 时序数据存储

位置: dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/sql/SqlInsertTsRepository.java

PostgreSQL 时序数据存储实现：

/**
 * 保存时序数据到 PostgreSQL
 */
@Override
public ListenableFuture<Integer> save(TenantId tenantId, EntityId entityId, 
                                     TsKvEntry tsKvEntry, long ttl) {
    // 1. 构建 SQL 插入语句
    String partition = getPartition(tsKvEntry.getTs());
    String tableName = getTableName(partition);
    
    // 2. 执行插入
    return executeAsync(() -> {
        return jdbcTemplate.update(
            "INSERT INTO " + tableName + " (entity_id, key, ts, bool_v, str_v, long_v, dbl_v, json_v) " +
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            entityId.getId(), 
            tsKvEntry.getKey(),
            tsKvEntry.getTs(),
            getBoolValue(tsKvEntry),
            getStrValue(tsKvEntry),
            getLongValue(tsKvEntry),
            getDblValue(tsKvEntry),
            getJsonValue(tsKvEntry)
        );
    });
}

3.4 TimescaleDB 时序数据存储

位置: dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/timescale/TimescaleInsertTsRepository.java

TimescaleDB 是 PostgreSQL 的时序数据扩展，提供更好的性能：

/**
 * 保存时序数据到 TimescaleDB
 * TimescaleDB 提供了 Hypertable 特性，自动管理分区
 */
@Override
public ListenableFuture<Integer> save(TenantId tenantId, EntityId entityId, 
                                     TsKvEntry tsKvEntry, long ttl) {
    // TimescaleDB 使用 Hypertable，不需要手动管理分区
    return executeAsync(() -> {
        return jdbcTemplate.update(
            "INSERT INTO ts_kv (entity_id, key, ts, bool_v, str_v, long_v, dbl_v, json_v) " +
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            entityId.getId(), 
            tsKvEntry.getKey(),
            tsKvEntry.getTs(),
            getBoolValue(tsKvEntry),
            getStrValue(tsKvEntry),
            getLongValue(tsKvEntry),
            getDblValue(tsKvEntry),
            getJsonValue(tsKvEntry)
        );
    });
}

3.5 Cassandra 时序数据存储

位置: dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraBufferedRateExecutor.java

Cassandra 提供高写入性能，适合大规模时序数据：

/**
 * 批量保存时序数据到 Cassandra
 */
public ListenableFuture<Void> save(TenantId tenantId, EntityId entityId, 
                                   List<TsKvEntry> entries, long ttl) {
    // 1. 构建 Cassandra 插入语句
    List<Statement> statements = new ArrayList<>();
    
    for (TsKvEntry entry : entries) {
        // 2. 构建分区键（entity_id + key + partition）
        String partition = getPartition(entry.getTs());
        
        // 3. 构建插入语句
        Insert insert = QueryBuilder.insertInto("ts_kv_cf")
            .value("entity_id", entityId.getId())
            .value("key", entry.getKey())
            .value("partition", partition)
            .value("ts", entry.getTs())
            .value("bool_v", getBoolValue(entry))
            .value("str_v", getStrValue(entry))
            .value("long_v", getLongValue(entry))
            .value("dbl_v", getDblValue(entry))
            .value("json_v", getJsonValue(entry));
        
        if (ttl > 0) {
            insert.using(QueryBuilder.ttl((int) ttl));
        }
        
        statements.add(insert);
    }
    
    // 4. 批量执行
    return executeAsync(() -> {
        session.execute(BatchStatement.newInstance(BatchStatement.Type.UNLOGGED, statements));
        return null;
    });
}

4. 数据分区策略

4.1 SQL 分区

PostgreSQL 使用表分区来管理时序数据：

/**
 * 获取分区名称
 * 分区策略：按月分区
 */
private String getPartition(long ts) {
    Calendar calendar = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
    calendar.setTimeInMillis(ts);
    return String.format("ts_kv_%d_%02d", 
        calendar.get(Calendar.YEAR), 
        calendar.get(Calendar.MONTH) + 1);
}

4.2 Cassandra 分区

Cassandra 使用分区键来分布数据：

/**
 * 获取 Cassandra 分区
 * 分区策略：按天分区
 */
private String getPartition(long ts) {
    return TimeUUIDs.timeUuid(ts).toString();
}

5. 最新值存储

5.1 最新值 DAO

位置: dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesLatestDao.java

最新值存储在独立的表中，便于快速查询：

/**
 * 保存最新值
 */
@Override
public ListenableFuture<Long> saveLatest(TenantId tenantId, EntityId entityId, 
                                         TsKvEntry tsKvEntry) {
    // 1. 构建 SQL UPDATE 语句（使用 ON CONFLICT 处理冲突）
    String sql = "INSERT INTO ts_kv_latest (entity_id, key, ts, bool_v, str_v, long_v, dbl_v, json_v) " +
                 "VALUES (?, ?, ?, ?, ?, ?, ?, ?) " +
                 "ON CONFLICT (entity_id, key) DO UPDATE SET " +
                 "ts = EXCLUDED.ts, " +
                 "bool_v = EXCLUDED.bool_v, " +
                 "str_v = EXCLUDED.str_v, " +
                 "long_v = EXCLUDED.long_v, " +
                 "dbl_v = EXCLUDED.dbl_v, " +
                 "json_v = EXCLUDED.json_v";
    
    // 2. 执行更新
    return executeAsync(() -> {
        return jdbcTemplate.update(sql, ...);
    });
}

6. 属性数据存储

6.1 属性服务

位置: application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java

属性数据包括：

SERVER_SCOPE: 服务器端属性
SHARED_SCOPE: 共享属性
CLIENT_SCOPE: 客户端属性

/**
 * 保存属性数据
 */
public void saveAttributes(AttributesSaveRequest request) {
    // 1. 验证权限
    // 2. 保存属性
    // 3. 发送更新通知
}

7. 数据查询

7.1 时序数据查询

位置: dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java

/**
 * 查询时序数据
 */
public ListenableFuture<List<TsKvEntry>> findAll(TenantId tenantId, 
                                                  EntityId entityId, 
                                                  List<ReadTsKvQuery> queries) {
    // 1. 构建查询任务列表
    List<ListenableFuture<List<TsKvEntry>>> futures = new ArrayList<>();
    
    for (ReadTsKvQuery query : queries) {
        // 2. 根据查询类型选择查询方法
        if (query.getAggregation() == Aggregation.NONE) {
            futures.add(timeseriesDao.findAll(tenantId, entityId, query));
        } else {
            futures.add(timeseriesDao.findAggregate(tenantId, entityId, query));
        }
    }
    
    // 3. 合并结果
    return Futures.transform(Futures.allAsList(futures), 
        results -> results.stream().flatMap(List::stream).collect(Collectors.toList()),
        MoreExecutors.directExecutor());
}

7.2 聚合查询

支持以下聚合函数：

AVG: 平均值
SUM: 求和
MIN: 最小值
MAX: 最大值
COUNT: 计数

8. 数据 TTL (Time To Live)

8.1 TTL 配置

数据可以配置 TTL，自动过期删除：

/**
 * 保存带 TTL 的数据
 */
public ListenableFuture<Integer> save(TenantId tenantId, EntityId entityId, 
                                     TsKvEntry tsKvEntry, long ttl) {
    if (ttl > 0) {
        // PostgreSQL: 使用 DELETE 触发器或定期清理
        // Cassandra: 使用 TTL 特性
        insert.using(QueryBuilder.ttl((int) ttl));
    }
}

9. 性能优化

9.1 批量插入

时序数据使用批量插入提高性能：

/**
 * 批量保存
 */
public ListenableFuture<Void> saveBatch(List<TsKvEntry> entries) {
    // 使用批量插入语句
    BatchStatement batch = BatchStatement.newInstance(...);
    for (TsKvEntry entry : entries) {
        batch.add(buildInsert(entry));
    }
    return executeAsync(() -> session.execute(batch));
}

9.2 异步处理

所有数据库操作都是异步的，使用 ListenableFuture：

/**
 * 异步执行数据库操作
 */
private <T> ListenableFuture<T> executeAsync(Callable<T> callable) {
    return executorService.submit(callable);
}

10. 总结

ThingsBoard 的数据存储系统具有以下特点：

分层设计: 清晰的 DAO 层抽象，支持多种数据库
多数据库支持: PostgreSQL、TimescaleDB、Cassandra
分区策略: SQL 按月分区，Cassandra 按天分区
最新值优化: 独立的最新值表，快速查询
异步处理: 所有操作异步执行，提高性能
TTL 支持: 自动数据过期清理
批量操作: 支持批量插入，提高写入性能

这套数据存储系统能够处理大规模的物联网时序数据，同时保持良好的查询性能。

14 KiB Raw Blame History Unescape Escape