使用Golang访问Apache IoTTDB时序数据库：环境搭建 + 连接池 + 全接口实战

# 启用 Go Modules
export GO111MODULE=on
# 配置代理
export GOPROXY=https://goproxy.io

# 创建项目目录
mkdir session_example && cd session_example

# 下载官方示例
curl -o session_example.go -L https://github.com/apache/iotdb-client-go/raw/main/example/session_example.go

# 初始化并下载依赖
go mod init session_example
go mod tidy

# 运行
go run session_example.go

# 安装 thrift
go get github.com/apache/thrift@0.13.0

# 创建目录
mkdir -p $GOPATH/src/iotdb-client-go-example/session_example
cd $GOPATH/src/iotdb-client-go-example/session_example

# 下载示例
curl -o session_example.go -L https://github.com/apache/iotdb-client-go/raw/main/example/session_example.go

# 初始化依赖
go mod init
go mod tidy

# 运行
go run session_example.go

config := &client.PoolConfig{
    Host:     "127.0.0.1",
    Port:     "6667",
    UserName: "root",
    Password: "root",
}
// 创建连接池：最大连接数3、连接超时60s、获取等待60s、关闭压缩
sessionPool = client.NewSessionPool(config, 3, 60000, 60000, false)
defer sessionPool.Close()

config := &client.PoolConfig{
    UserName: "root",
    Password: "root",
    NodeUrls: strings.Split("127.0.0.1:6667,127.0.0.1:6668", ","),
}
sessionPool = client.NewSessionPool(config, 3, 60000, 60000, false)
defer sessionPool.Close()

session, err := sessionPool.GetSession()
defer sessionPool.PutBack(session)

status, err := session.InsertTablet(tablet, false)
tablet.Reset()

var timeout int64 = 1000
session, err := sessionPool.GetSession()
defer sessionPool.PutBack(session)

sessionDataSet, err := session.ExecuteQueryStatement(sql, &timeout)
defer sessionDataSet.Close()

package main

import (
    "flag"
    "fmt"
    "log"
    "math/rand"
    "strings"
    "time"
    "github.com/apache/iotdb-client-go/v2/client"
    "github.com/apache/iotdb-client-go/v2/common"
)

var (
    host     string
    port     string
    user     string
    password string
    sessionPool client.SessionPool
)

func main() {
    flag.StringVar(&host, "host", "127.0.0.1", "--host=192.168.1.100")
    flag.StringVar(&port, "port", "6667", "--port=6667")
    flag.StringVar(&user, "user", "root", "--user=root")
    flag.StringVar(&password, "password", "root", "--password=root")
    flag.Parse()

    // 1. 创建连接池
    config := &client.PoolConfig{Host: host, Port: port, UserName: user, Password: password}
    sessionPool = client.NewSessionPool(config, 3, 60000, 60000, false)
    defer sessionPool.Close()

    // 2. 元数据操作
    setStorageGroup("root.sg1")
    createTimeseries("root.sg1.dev1.temperature")

    // 3. 写入数据
    insertTablet()

    // 4. 查询数据
    executeQueryStatement("select temperature from root.sg1.dev1")

    // 5. 清理资源
    deleteTimeseries("root.sg1.dev1.temperature")
    deleteStorageGroup("root.sg1")
}

// 设置存储组
func setStorageGroup(sg string) {
    session, err := sessionPool.GetSession()
    defer sessionPool.PutBack(session)
    if err == nil {
        session.SetStorageGroup(sg)
    }
}

// 创建时间序列
func createTimeseries(path string) {
    session, err := sessionPool.GetSession()
    defer sessionPool.PutBack(session)
    if err == nil {
        checkError(session.CreateTimeseries(path, client.FLOAT, client.PLAIN, client.SNAPPY, nil, nil))
    }
}

// 插入Tablet
func insertTablet() {
    session, err := sessionPool.GetSession()
    defer sessionPool.PutBack(session)
    if err == nil {
        tablet, err := client.NewTablet("root.sg1.dev1", []*client.MeasurementSchema{{Measurement: "temperature", DataType: client.FLOAT}}, 12)
        if err != nil {
            log.Fatal(err)
        }
        ts := time.Now().UTC().UnixNano() / 1000000
        for row := 0; row < 12; row++ {
            ts++
            tablet.SetTimestamp(ts, row)
            tablet.SetValueAt(rand.Float32(), 0, row)
            tablet.RowSize++
        }
        status, err := session.InsertTablet(tablet, false)
        tablet.Reset()
        checkError(status, err)
    }
}

// 查询语句
func executeQueryStatement(sql string) {
    var timeout int64 = 1000
    session, err := sessionPool.GetSession()
    defer sessionPool.PutBack(session)
    if err != nil {
        log.Print(err)
        return
    }
    sds, err := session.ExecuteQueryStatement(sql, &timeout)
    if err == nil {
        defer sds.Close()
        cols := sds.GetColumnNames()
        for _, c := range cols { fmt.Printf("%s\t", c) }
        fmt.Println()
        for next, _ := sds.Next(); next; next, _ = sds.Next() {
            for _, c := range cols {
                if null, _ := sds.IsNull(c); null {
                    fmt.Print("null\t")
                } else {
                    v, _ := sds.GetString(c)
                    fmt.Printf("%s\t", v)
                }
            }
            fmt.Println()
        }
    }
}

// 删除时序
func deleteTimeseries(paths ...string) {
    session, err := sessionPool.GetSession()
    defer sessionPool.PutBack(session)
    if err == nil { checkError(session.DeleteTimeseries(paths)) }
}

// 删除存储组
func deleteStorageGroup(sg string) {
    session, err := sessionPool.GetSession()
    defer sessionPool.PutBack(session)
    if err == nil { checkError(session.DeleteStorageGroup(sg)) }
}

// 错误检查
func checkError(status *common.TSStatus, err error) {
    if err != nil { log.Fatal(err) }
    if status != nil {
        if e := client.VerifySuccess(status); e != nil { log.Println(e) }
    }
}