InfluxDB 时序数据的高效解决方案

docker run \
    --name alex-influxdb \
    -p 8086:8086 \
    influxdb:2

# $PWD/data：要挂载在容器的 InfluxDB 数据目录路径下的主机目录
# $PWD/config：要挂载到容器的 InfluxDB 配置目录路径下的主机目录
# <USERNAME>：初始管理员用户的名称
# <PASSWORD>： 初始管理员用户的密码
# <ORG_NAME>：初始组织的名称
# <BUCKET_NAME>：初始存储桶（数据库）的名称
docker run -d -p 8086:8086 \
  -v "$PWD/data:/var/lib/influxdb2" \
  -v "$PWD/config:/etc/influxdb2" \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=<USERNAME> \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=<PASSWORD> \
  -e DOCKER_INFLUXDB_INIT_ORG=<ORG_NAME> \
  -e DOCKER_INFLUXDB_INIT_BUCKET=<BUCKET_NAME> \
  influxdb:2

go get -v github.com/influxdata/influxdb-client-go/v2

package main

import (
	"context"
	"crypto/rand"
	"fmt"
	"math/big"
	"strconv"
	"time"

	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
)

// 配置信息
const (
	serverURL   = "http://127.0.0.1:8086" // InfluxDB 地址
	token       = "your_token"            // 认证令牌（需替换为实际值）
	org         = "your_org"              // 组织名称（需替换）
	bucket      = "your_bucket"           // 存储桶名称（需替换）
	measurement = "sensor_data"           // 表名（Measurement）
)

func connInflux() influxdb2.Client {
	// 1. 创建 InfluxDB 客户端
	client := influxdb2.NewClient(serverURL, token)

	return client
}

// GenerateRandomDigit 生成一个 1 到 9 之间的随机数字
func GenerateRandomDigit() int {
    num, err := rand.Int(rand.Reader, big.NewInt(9))
    if err != nil {
        return 1
    }
    return int(num.Int64()) + 1
}

// GenerateRandomNumber 生成指定长度的随机数字，并转换为整型
func GenerateRandomNumber(length int) (int, error) {
    const digits = "0123456789"
    result := make([]byte, length)
    for i := 0; i < length; i++ {
        num, err := rand.Int(rand.Reader, big.NewInt(int64(len(digits))))
        if err != nil {
            return 0, err
        }
        result[i] = digits[num.Int64()]
    }
    numInt, err := strconv.Atoi(string(result))
    if err != nil {
        return 0, err
    }
    return numInt, nil
}

// writePoints 将一条数据写入 InfluxDB
func writePoints(client influxdb2.Client) {
    // 获取写入 API，使用阻塞方式保证数据写入成功后再继续执行
    writeAPI := client.WriteAPIBlocking(org, bucket)

    // 随机生成 temperature 和 humidity 数据
    t1, err := GenerateRandomNumber(GenerateRandomDigit())
    if err != nil {
        panic(fmt.Sprintf("生成随机数 temperature 失败: %v", err))
    }
    t2, err := GenerateRandomNumber(GenerateRandomDigit())
    if err != nil {
        panic(fmt.Sprintf("生成随机数 humidity 失败: %v", err))
    }

    // ========== C (Create)：写入数据 ==========
    // 创建一个数据点（Point）
    p := influxdb2.NewPoint(
        measurement,
        // Tags：用于快速过滤查询，类似于 MySQL 的索引字段
        map[string]string{"device_id": "d1", "location": "room1"},
        // Fields：存储实际数据值，这里 temperature 使用浮点数，humidity 使用整型
        map[string]interface{}{"temperature": float64(t1), "humidity": t2},
        // 当前时间作为时间戳
        time.Now(),
    )

    // 写入数据点到 InfluxDB
    if err := writeAPI.WritePoint(context.Background(), p); err != nil {
        panic(fmt.Sprintf("写入失败: %v", err))
    }
    fmt.Println("数据写入成功！")
}

// fetchData 从 InfluxDB 查询数据
func fetchData(client influxdb2.Client) {
    queryAPI := client.QueryAPI(org)

    // ========== R (Read)：查询数据 ==========
    // 构建 Flux 查询语句，查询最近 1 小时内，且 device_id 为 "d1" 的数据
    query := fmt.Sprintf(`from(bucket: "%s")
        |> range(start: -1h)
        |> filter(fn: (r) => r._measurement == "%s" and r.device_id == "d1")`, bucket, measurement)

    // 执行查询
    result, err := queryAPI.Query(context.Background(), query)
    if err != nil {
        panic(fmt.Sprintf("查询失败: %v", err))
    }

    // 遍历查询结果并打印每一条记录
    fmt.Println("\n查询结果:")
    for result.Next() {
        record := result.Record()
        fmt.Printf(
            "[ %s ] %s : %v\n",
            record.Time().Format(time.RFC3339), // 格式化时间
            record.Field(),                     // 字段名（例如 temperature）
            record.Value(),                     // 字段值
        )
        fmt.Println()
    }
    if result.Err() != nil {
        panic(fmt.Sprintf("结果解析错误: %v", result.Err()))
    }
}

func main() {
	client := connInflux()
	defer client.Close() // 确保最后关闭客户端

	// 尝试每隔 500ms 写入一次数据，写 100 次
	// for i := 0; i < 100; i++ {
	// 	writePoints(client)
	// 	time.Sleep(500 * time.Millisecond)
	// }

	// 查询数据
	fetchData(client)
}