Flutter数据监控与分析插件influxdb_client的使用

概述

influxdb_client 是一个用于与 InfluxDB 2.x 交互的 Dart 客户端库。它支持在 Web、服务器和 Flutter 应用程序中使用。本文将详细介绍如何安装和使用 influxdb_client 插件，包括创建客户端、写入数据、查询数据、删除数据以及管理 API 的使用。

安装

在 pubspec.yaml 文件中添加 influxdb_client 依赖：

dependencies:
  influxdb_client: ^2.2.0

然后在 Dart 文件中导入库：

import 'package:influxdb_client/api.dart';

使用

创建客户端

创建 InfluxDBClient 实例时，需要指定 InfluxDB 的 URL 和 Token。还可以设置默认的组织（org）和存储桶（bucket），以及其他选项如调试模式等。

var client = InfluxDBClient(
    url: 'http://localhost:8086',
    token: 'my-token',
    org: 'my-org',
    bucket: 'my-bucket',
    debug: true);

写入数据

WriteApi 支持异步写入数据到 InfluxDB 2.x。数据可以是字符串形式的 Line Protocol，也可以是 Point 结构体，或者两者的数组。

示例代码

import 'dart:async';
import 'dart:math';
import 'package:influxdb_client/api.dart';

void main() async {
  // 创建 InfluxDBClient
  var client = InfluxDBClient(
      url: 'http://localhost:8086',
      token: 'my-token',
      org: 'my-org',
      bucket: 'my-bucket',
      debug: true);

  // 创建写服务
  var writeApi = client.getWriteService(WriteOptions().merge(
      precision: WritePrecision.s,
      batchSize: 100,
      flushInterval: 5000,
      gzip: true));

  // 创建数据点列表
  var data = List<Point>.empty(growable: true);
  var random = Random();
  for (var i = 0; i < 10; i++) {
    var temperature = random.nextInt(30);
    data.add(Point('weather')
        .addTag('location', 'Prague')
        .addField('temperature', temperature)
        .time(DateTime.now().subtract(Duration(days: i)).toUtc()));
  }

  // 写入数据到 InfluxDB
  print(
      '\n\n-------------------------------- Write data -------------------------------\n');
  await writeApi.write(data).then((value) {
    print('Write completed 1');
  }).catchError((exception) {
    print('Handle write error here!');
    print(exception);
  });
}

查询数据

QueryService 可以用于从 InfluxDB 查询数据。查询结果可以格式化为 FluxRecord 或者原始字符串。

示例代码

// 创建查询服务并执行查询
var queryService = client.getQueryService();
var fluxQuery = '''
       from(bucket: "my-bucket")
            |> range(start: -20d)
            |> filter(fn: (r) => r["_measurement"] == "weather" 
                             and r["location"] == "Prague")''';

// 查询并迭代所有记录
print(
    '\n\n---------------------------------- Query ---------------------------------\n');
var recordStream = await queryService.query(fluxQuery);

print(
    '\n\n------------------------------ Query result ------------------------------\n');
await recordStream.forEach((record) {
  print(
      'Temperature in ${record['location']} at ${record['_time']} is ${record['_value']} °C');
});

删除数据

DeleteService 可以用于从 InfluxDB 删除数据。删除操作需要指定条件、开始时间、结束时间、存储桶和组织。

示例代码

// 删除数据
print(
    '\n\n------------------------------- Delete data -------------------------------\n');
await client
    .getDeleteService()
    .delete(
        predicate: '_measurement="weather"',
        start: DateTime.parse('1970-01-01T00:00:00Z'),
        stop: DateTime.now().toUtc(),
        bucket: 'my-bucket',
        org: 'my-org')
    .catchError((e) => print(e));

管理 API

influxdb_client 还提供了管理 API，可以用于创建、删除存储桶、授权等操作。

示例代码

import 'package:influxdb_client/api.dart';

void main() async {
  // 初始化客户端和 API
  var client = InfluxDBClient(
      url: 'http://localhost:8086', token: 'my-token', org: 'my-org');

  // 检查服务器可用性
  await client.getPingApi().getPing();

  var orgs = await client.getOrganizationsApi().getOrgs();
  var myOrgId = orgs.orgs.first.id;

  var bucketsApi = client.getBucketsApi();
  var bucketName = 'bucket-my-org';

  // 查找并删除名为 'bucket-my-org' 的存储桶
  var buckets = await bucketsApi.getBuckets(name: bucketName);
  if (buckets.buckets.isNotEmpty) {
    var bucketID = buckets.buckets.first.id;
    await bucketsApi.deleteBucketsID(bucketID);
    print('Bucket $bucketID was deleted.');
  }

  // 存储桶配置
  var request = PostBucketRequest(
      orgID: myOrgId,
      name: bucketName,
      retentionRules: [
        RetentionRule(type: RetentionRuleTypeEnum.expire, everySeconds: 3600)
      ]);

  var bucket = await bucketsApi.postBuckets(request);

  // 创建授权，允许读写新创建的存储桶
  var bucketResource =
      Resource(type: ResourceTypeEnum.buckets, id: bucket.id, orgID: myOrgId);

  // 授权配置
  var auth = AuthorizationPostRequest(
      description: 'Authorization to read/write bucket:${bucket.name}',
      orgID: myOrgId,
      permissions: [
        Permission(action: PermissionActionEnum.read, resource: bucketResource),
        Permission(action: PermissionActionEnum.write, resource: bucketResource)
      ]);

  // 创建授权
  var authorizationsApi = client.getAuthorizationsApi();
  var authorization = await authorizationsApi.postAuthorizations(auth);

  // 打印 Token
  var token = authorization.token;
  print('The bucket: \'${bucket.name}\' is successfully created.');
  print('The following token can be used to read/write: ${token}');

  client.close();
}

总结

influxdb_client 是一个功能强大的 Dart 客户端库，适用于与 InfluxDB 2.x 进行数据交互。通过本文的介绍，您应该能够轻松地在 Flutter 应用程序中集成 influxdb_client，实现数据的写入、查询和删除等操作。希望这些示例代码对您有所帮助！