【物联网】时序数据库InfluxDB解析及1.x版本与2.x版本区别详解

一、什么是时序数据库？

时序数据库（Time-Series Database，TSDB）是一种专门为处理时间序列数据而设计的数据库。时间序列数据是指按时间顺序排列的数据，通常用于记录和查询系统、传感器、服务器等在一段时间内的运行状态。时序数据库具有以下特点：

1. 数据模型简单：通常包含时间戳、度量值和可选的标签。
2. 高效的写入性能：能够快速处理大量数据写入。
3. 高效的查询性能：针对时间范围、聚合等操作进行优化。
4. 数据压缩：通过数据压缩降低存储成本。

二、InfluxDB简介

InfluxDB是一款开源的时序数据库，适用于存储和分析大规模的时间序列数据。它具有以下优势：

1. 易于安装和使用：InfluxDB安装简单，提供友好的HTTP API接口。
2. 高性能：针对时序数据的写入和查询进行了优化。
3. 可扩展性：支持集群部署，可根据需求扩展节点。
4. 丰富的功能：支持数据保留策略、连续查询等。

三、InfluxDB 1.x版本与2.x版本区别

1. 架构变化：InfluxDB 2.x版本在架构上进行了优化，将原先的多个组件（InfluxDB、InfluxDB OSS、InfluxDB Enterprise）整合为一个统一的InfluxDB 2.x版本。
2. 数据模型：InfluxDB 2.x版本对数据模型进行了简化，将measurement、tag、field等概念统一为bucket。
3. API升级：InfluxDB 2.x版本提供了全新的Flux查询语言，同时兼容InfluxQL。
4. 安全性：InfluxDB 2.x版本加强了安全性，默认启用身份验证和授权。
5. 可视化：InfluxDB 2.x版本内置了可视化工具，方便用户进行数据分析和展示。
6. 集群管理：InfluxDB 2.x版本简化了集群管理，支持自动故障转移。

四、代码实现数据插入与查询

以下是基于InfluxDB 1.x版本和2.x版本的Java代码示例，实现数据插入与查询。

1. InfluxDB 1.x版本
   （1）数据插入

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Point;
public class InfluxDB1xExample {
    public static void main(String[] args) {
        InfluxDB influxDB = InfluxDBFactory.connect("http://localhost:8086", "username", "password");
        influxDB.setDatabase("mydb");
        Point point = Point.measurement("cpu_usage")
                .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
                .tag("region", "us-west")
                .addField("value", 0.64)
                .build();
        influxDB.write(point);
    }
}

（2）数据查询

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
public class InfluxDB1xExample {
    public static void main(String[] args) {
        InfluxDB influxDB = InfluxDBFactory.connect("http://localhost:8086", "username", "password");
        String query = "SELECT * FROM cpu_usage WHERE region='us-west' LIMIT 10";
        QueryResult results = influxDB.query(new Query(query, "mydb"));
        for (QueryResult.Result result : results.getResults()) {
            for (QueryResult.Series series : result.getSeries()) {
                System.out.println(series.toString());
            }
        }
    }
}

1. InfluxDB 2.x版本
   （1）数据插入

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import com.influxdb.client.WriteApi;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
public class InfluxDB2xExample {
    public static void main(String[] args) {
        InfluxDBClient influxDBClient = InfluxDBClientFactory.create("http://localhost:8086", "mytoken".toCharArray());
        WriteApi writeApi = influxDBClient.getWriteApi();
        Point point = Point.measurement("cpu_usage")
                .addTag("region", "us-west")
                .addField("value", 0.64)
                .time(System.currentTimeMillis(), WritePrecision.MS);
        writeApi.writePoint(point);
        writeApi.close();
        influxDBClient.close();
    }
}

（2）数据查询

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import com.influxdb.client.QueryApi;
import com.influxdb.client.domain.Dialect;
import com.influxdb.client.domain.Query;
import com.influxdb.client.domain.QueryParam;
public class InfluxDB2xExample {
    public static void main(String[] args) {
        InfluxDBClient influxDBClient = InfluxDBClientFactory.create("http://localhost:8086", "mytoken".toCharArray());
        QueryApi queryApi = influxDBClient.getQueryApi();
        String fluxQuery = "from(bucket: \"mybucket\") |> range(start: -1h) |> filter(fn: (r) => r._measurement == \"cpu_usage\" and r.region == \"us-west\") |> limit(n: 10)";
        Query query = new Query(fluxQuery, "myorg");
        queryApi.query(query, 20, new Consumer<FluxTable>() {
            @Override
            public void accept(FluxTable fluxTable) {
                fluxTable.getRecords().forEach(record -> {
                    System.out.println(record.getTime() + ": " + record.getValueByKey("_value"));
                });
            }
        });
        influxDBClient.close();
    }
}

在上面的代码示例中，我们首先创建了一个InfluxDBClient实例，这是InfluxDB 2.x版本提供的客户端库。然后，我们使用WriteApi来写入数据点，使用QueryApi来执行Flux查询语言编写的查询。
请注意，InfluxDB 2.x版本的API与1.x版本有很大的不同，特别是在查询语言方面。InfluxDB 2.x版本引入了Flux，这是一种更强大、更灵活的查询语言，用于处理和分析时序数据。
在2.x版本的查询示例中，我们使用了Flux查询来从特定的bucket中选择数据，并应用了一系列的操作，如range（选择时间范围）、filter（过滤数据）和limit（限制结果数量）。
最后，我们通过queryApi.query方法执行查询，并使用一个Consumer来处理查询结果。在这个例子中，我们简单地打印出每个记录的时间戳和值。
总结来说，InfluxDB 2.x版本在API和功能上都有显著的改进，提供了更现代的编程模型和更丰富的查询能力。开发者在选择使用哪个版本时，需要根据自己的项目需求和兼容性考虑。