如何使用golang客户端连续写入influxdb答案

【问题标题】：How to write to continuosly write to influxdb using golang client如何使用golang客户端连续写入influxdb
【发布时间】：2016-10-30 06:18:46
【问题描述】：

我正在使用 influxDB 来存储我的时间序列数据。

我编写了一个简单的 golang 应用程序来从名为 time.log 的文件中读取行。

https://github.com/influxdata/influxdb/blob/master/client/README.md#inserting-data 的文档说：

插入数据

使用批量插入将时间序列数据（即点）写入数据库。该机制是创建一个或多个点，然后创建一个批次（也称为批次点）并将这些写入给定的数据库和系列。系列是测量（时间/值）和一组标签的组合。

在此示例中，我们将创建一个包含 1,000 个点的批次。每个点都有一个时间和一个值以及 2 个指示形状和颜色的标签。我们使用名为 shape 的度量将这些点写入名为 square_holes 的数据库。

注意：您可以将 RetentionPolicy 指定为批处理点的一部分。如果未提供，InfluxDB 将使用数据库默认保留策略。
func writePoints(clnt client.Client) {
    sampleSize := 1000
    rand.Seed(42)

    bp, _ := client.NewBatchPoints(client.BatchPointsConfig{
        Database:  "systemstats",
        Precision: "us",
    })

    for i := 0; i < sampleSize; i++ {
        regions := []string{"us-west1", "us-west2", "us-west3", "us-east1"}
        tags := map[string]string{
            "cpu":    "cpu-total",
            "host":   fmt.Sprintf("host%d", rand.Intn(1000)),
            "region": regions[rand.Intn(len(regions))],
        }

        idle := rand.Float64() * 100.0
        fields := map[string]interface{}{
            "idle": idle,
            "busy": 100.0 - idle,
        }

        bp.AddPoint(client.NewPoint(
            "cpu_usage",
            tags,
            fields,
            time.Now(),
        ))
    }

    err := clnt.Write(bp)
    if err != nil {
        log.Fatal(err)
    }
}

但是因为我不断地从日志中读取数据。我从来没有读完日志。那么我将积分写入流入服务器的最佳方式是什么？

这是我当前的代码：

cmdBP := client.NewBatchPoints(...)
for line := range logFile.Lines {
    pt := parseLine(line.Text)
    cmdBP.AddPoint(pt)
}

influxClient.Write(cmdBP)

基本上范围 logFile.Lines 永远不会终止，因为它基于通道。

【问题讨论】：

每处理 N 行日志就调用 client.Write(cmdBP) 怎么样？
所以我需要在范围循环内使用计数器？

标签： logging go influxdb

【解决方案1】：

使用批处理点和超时的组合（作为 goroutine 运行）：

func (h *InfluxDBHook) loop() {
    var coll []*client.Point
    tick := time.NewTicker(h._batchInterval)

    for {
        timeout := false

        select {
        case pt := <-h._points:
            coll = append(coll, pt)
        case <-tick.C:
            timeout = true
        }

        if (timeout || len(coll) >= h._batchSize) && len(coll) > 0 {
            bp, err := client.NewBatchPoints(h._batchPointsConfig)
            if err != nil {
                //TODO:
            }
            bp.AddPoints(coll)
            err = h._client.Write(bp)
            if err != nil {
                //TODO:
            } else {
                coll = nil
            }
        }
    }
}

顺便说一句，您可以使用带有 logrus 日志记录包的挂钩，将日志发送到 InfluxDB（示例代码来自 logrus InfluxDB hook）。

【讨论】：