TDengine Go Connector
driver-go
是 TDengine 的官方 Go 语言连接器,实现了 Go 语言 database/sql 包的接口。Go 开发人员可以通过它开发存取 TDengine 集群数据的应用软件。
连接方式
driver-go
提供三种建立连接的方式。
- 原生连接,通过 TDengine 客户端驱动程序(taosc)原生连接 TDengine 实例,支持数据写入、查询、数据订阅、schemaless 接口和参数绑定接口等功能。
- REST 连接,通过 taosAdapter 提供的 HTTP 接口连接 TDengine 实例,不支持 schemaless 和数据订阅等特性。
- Websocket 连接,通过 taosAdapter 提供的 Websocket 接口连接 TDengine 实例,WebSocket 连接实现的功能集合和原生连接有少量不同。
连接方式的详细介绍请参考:连接器建立连接的方式
兼容性
支持最低 Go 版本 1.14,建议使用最新 Go 版本
支持的平台
原生连接支持的平台和 TDengine 客户端驱动支持的平台一致。 REST 连接支持所有能运行 Go 的平台。
版本支持
请参考版本支持列表
处理异常
如果是 TDengine 错误可以通过以下方式获取错误码和错误信息。
// import "github.com/taosdata/driver-go/v3/errors"
if err != nil {
tError, is := err.(*errors.TaosError)
if is {
fmt.Println("errorCode:", int(tError.Code))
fmt.Println("errorMessage:", tError.ErrStr)
} else {
fmt.Println(err.Error())
}
}
TDengine DataType 和 Go DataType
TDengine DataType | Go Type |
---|---|
TIMESTAMP | time.Time |
TINYINT | int8 |
SMALLINT | int16 |
INT | int32 |
BIGINT | int64 |
TINYINT UNSIGNED | uint8 |
SMALLINT UNSIGNED | uint16 |
INT UNSIGNED | uint32 |
BIGINT UNSIGNED | uint64 |
FLOAT | float32 |
DOUBLE | float64 |
BOOL | bool |
BINARY | string |
NCHAR | string |
JSON | []byte |
注意:JSON 类型仅在 tag 中支持。
安装步骤
安装前准备
- 安装 Go 开发环境(Go 1.14 及以上,GCC 4.8.5 及以上)
- 如果使用原生连接器,请安装 TDengine 客户端驱动,具体步骤请参考安装客户端驱动
配置好环境变量,检查命令:
go env
gcc -v
安装连接器
-
使用
go mod
命令初始化项目:go mod init taos-demo
-
引入 taosSql :
import (
"database/sql"
_ "github.com/taosdata/driver-go/v3/taosSql"
) -
使用
go mod tidy
更新依赖包:go mod tidy
-
使用
go run taos-demo
运行程序或使用go build
命令编译出二进制文件。
go run taos-demo
go build
建立连接
数据源名称具有通用格式,例如 PEAR DB,但没有类型前缀(方括号表示可选):
[username[:password]@][protocol[(address)]]/[dbname][?param1=value1&...¶mN=valueN]
完整形式的 DSN:
username:password@protocol(address)/dbname?param=value
- 原生连接
- REST 连接
- WebSocket 连接
taosSql 通过 cgo 实现了 Go 的 database/sql/driver
接口。只需要引入驱动就可以使用 database/sql
的接口。
使用 taosSql
作为 driverName
并且使用一个正确的 DSN 作为 dataSourceName
,DSN 支持的参数:
- cfg 指定 taos.cfg 目录
示例:
package main
import (
"database/sql"
"fmt"
_ "github.com/taosdata/driver-go/v3/taosSql"
)
func main() {
var taosUri = "root:taosdata@tcp(localhost:6030)/"
taos, err := sql.Open("taosSql", taosUri)
if err != nil {
fmt.Println("failed to connect TDengine, err:", err)
return
}
}
taosRestful 通过 http client
实现了 Go 的 database/sql/driver
接口。只需要引入驱动就可以使用database/sql
的接口。
使用 taosRestful
作为 driverName
并且使用一个正确的 DSN 作为 dataSourceName
,DSN 支持的参数:
disableCompression
是否接受压缩数据,默认为 true 不接受压缩数据,如果传输数据使用 gzip 压缩设置为 false。readBufferSize
读取数据的缓存区大小默认为 4K(4096),当查询结果数据量多时可以适当调大该值。
示例:
package main
import (
"database/sql"
"fmt"
_ "github.com/taosdata/driver-go/v3/taosRestful"
)
func main() {
var taosUri = "root:taosdata@http(localhost:6041)/"
taos, err := sql.Open("taosRestful", taosUri)
if err != nil {
fmt.Println("failed to connect TDengine, err:", err)
return
}
}
taosWS 通过 WebSocket
实现了 Go 的 database/sql/driver
接口。只需要引入驱动(driver-go 最低版本 3.0.2)就可以使用database/sql
的接口。
使用 taosWS
作为 driverName
并且使用一个正确的 DSN 作为 dataSourceName
,DSN 支持的参数:
writeTimeout
通过 WebSocket 发送数据的超时时间。readTimeout
通过 WebSocket 接收响应数据的超时时间。
示例:
package main
import (
"database/sql"
"fmt"
_ "github.com/taosdata/driver-go/v3/taosWS"
)
func main() {
var taosUri = "root:taosdata@ws(localhost:6041)/"
taos, err := sql.Open("taosWS", taosUri)
if err != nil {
fmt.Println("failed to connect TDengine, err:", err)
return
}
}
指定 URL 和 Properties 获取连接
Go 连接器不支持此功能
配置参数的优先级
Go 连接器不支持此功能
使用示例
创建数据库和表
_, err = taos.Exec("CREATE DATABASE if not exists power")
if err != nil {
log.Fatalln("failed to create database, err:", err)
}
_, err = taos.Exec("CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))")
if err != nil {
log.Fatalln("failed to create stable, err:", err)
}
插入数据
affected, err := taos.Exec("INSERT INTO " +
"power.d1001 USING power.meters TAGS(2,'California.SanFrancisco') " +
"VALUES " +
"(NOW + 1a, 10.30000, 219, 0.31000) " +
"(NOW + 2a, 12.60000, 218, 0.33000) " +
"(NOW + 3a, 12.30000, 221, 0.31000) " +
"power.d1002 USING power.meters TAGS(3, 'California.SanFrancisco') " +
"VALUES " +
"(NOW + 1a, 10.30000, 218, 0.25000) ")
if err != nil {
log.Fatalln("failed to insert data, err:", err)
}
log.Println("affected rows:", affected)
查询数据
rows, err := taos.Query("SELECT * FROM power.meters")
if err != nil {
log.Fatalln("failed to select from table, err:", err)
}
defer rows.Close()
for rows.Next() {
var (
ts time.Time
current float32
voltage int
phase float32
groupId int
location string
)
err := rows.Scan(&ts, ¤t, &voltage, &phase, &groupId, &location)
if err != nil {
log.Fatalln("scan error:\n", err)
return
}
log.Println(ts, current, voltage, phase, groupId, location)
}
执行带有 reqId 的 SQL
reqId 可用于请求链路追踪,reqId 就像分布式系统中的 traceId 作用一样。一个请求可能需要经过多个服务或者模块才能完成。reqId 用于标识和关联这个请求的所有相关操作,以便于我们可以追踪和分析请求的完整路径。
使用 reqId 有下面好处:
- 请求追踪:通过将同一个 reqId 关联到一个请求的所有相关操作,可以追踪请求在系统中的完整路径
- 性能分析:通过分析一个请求的 reqId,可以了解请求在各个服务和模块中的处理时间,从而找出性能瓶颈
- 故障诊断:当一个请求失败时,可以通过查看与该请求关联的 reqId 来找出问题发生的位置
如果用户不设置reqId,连接器也会内部随机生成一个,但是还是建议用户设置,可以更好的跟用户请求关联起来。
ctx := context.WithValue(context.Background(), common.ReqIDKey, common.GetReqID())
_, err = taos.ExecContext(ctx, "CREATE DATABASE IF NOT EXISTS power")
if err != nil {
log.Fatalln("failed to create database, err:", err)
}
通过参数绑定写入数据
- 原生连接
- WebSocket 连接
package main
import (
"fmt"
"strconv"
"time"
"github.com/taosdata/driver-go/v3/af"
"github.com/taosdata/driver-go/v3/common"
"github.com/taosdata/driver-go/v3/common/param"
)
const (
NumOfSubTable = 10
NumOfRow = 10
)
func main() {
prepare()
db, err := af.Open("", "root", "taosdata", "power", 0)
if err != nil {
panic(err)
}
defer db.Close()
stmt := db.InsertStmt()
defer stmt.Close()
err = stmt.Prepare("INSERT INTO ? USING meters TAGS(?,?) VALUES (?,?,?,?)")
if err != nil {
panic(err)
}
for i := 1; i <= NumOfSubTable; i++ {
tags := param.NewParam(2).AddInt(i).AddBinary([]byte("location"))
err = stmt.SetTableNameWithTags("d_bind_"+strconv.Itoa(i), tags)
if err != nil {
panic(err)
}
now := time.Now()
params := make([]*param.Param, 4)
params[0] = param.NewParam(NumOfRow)
params[1] = param.NewParam(NumOfRow)
params[2] = param.NewParam(NumOfRow)
params[3] = param.NewParam(NumOfRow)
for i := 0; i < NumOfRow; i++ {
params[0].SetTimestamp(i, now.Add(time.Duration(i)*time.Second), common.PrecisionMilliSecond)
params[1].SetFloat(i, float32(i))
params[2].SetInt(i, i)
params[3].SetFloat(i, float32(i))
}
paramTypes := param.NewColumnType(4).AddTimestamp().AddFloat().AddInt().AddFloat()
err = stmt.BindParam(params, paramTypes)
if err != nil {
panic(err)
}
err = stmt.AddBatch()
if err != nil {
panic(err)
}
err = stmt.Execute()
if err != nil {
panic(err)
}
affected := stmt.GetAffectedRows()
fmt.Println("affected rows:", affected)
}
}
func prepare() {
db, err := af.Open("", "root", "taosdata", "", 0)
if err != nil {
panic(err)
}
defer db.Close()
_, err = db.Exec("CREATE DATABASE IF NOT EXISTS power")
if err != nil {
panic(err)
}
_, err = db.Exec("CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))")
if err != nil {
panic(err)
}
}
package main
import (
"database/sql"
"fmt"
"strconv"
"time"
"github.com/taosdata/driver-go/v3/common"
"github.com/taosdata/driver-go/v3/common/param"
_ "github.com/taosdata/driver-go/v3/taosRestful"
"github.com/taosdata/driver-go/v3/ws/stmt"
)
const (
NumOfSubTable = 10
NumOfRow = 10
)
func main() {
prepare()
config := stmt.NewConfig("ws://127.0.0.1:6041", 0)
config.SetConnectUser("root")
config.SetConnectPass("taosdata")
config.SetConnectDB("power")
config.SetMessageTimeout(common.DefaultMessageTimeout)
config.SetWriteWait(common.DefaultWriteWait)
config.SetErrorHandler(func(connector *stmt.Connector, err error) {
panic(err)
})
config.SetCloseHandler(func() {
fmt.Println("stmt connector closed")
})
connector, err := stmt.NewConnector(config)
if err != nil {
panic(err)
}
stmt, err := connector.Init()
err = stmt.Prepare("INSERT INTO ? USING meters TAGS(?,?) VALUES (?,?,?,?)")
if err != nil {
panic(err)
}
for i := 1; i <= NumOfSubTable; i++ {
tags := param.NewParam(2).AddInt(i).AddBinary([]byte("location"))
err = stmt.SetTableName("d_bind_" + strconv.Itoa(i))
if err != nil {
panic(err)
}
err = stmt.SetTags(tags, param.NewColumnType(2).AddInt().AddBinary(8))
now := time.Now()
params := make([]*param.Param, 4)
params[0] = param.NewParam(NumOfRow)
params[1] = param.NewParam(NumOfRow)
params[2] = param.NewParam(NumOfRow)
params[3] = param.NewParam(NumOfRow)
for i := 0; i < NumOfRow; i++ {
params[0].SetTimestamp(i, now.Add(time.Duration(i)*time.Second), common.PrecisionMilliSecond)
params[1].SetFloat(i, float32(i))
params[2].SetInt(i, i)
params[3].SetFloat(i, float32(i))
}
paramTypes := param.NewColumnType(4).AddTimestamp().AddFloat().AddInt().AddFloat()
err = stmt.BindParam(params, paramTypes)
if err != nil {
panic(err)
}
err = stmt.AddBatch()
if err != nil {
panic(err)
}
err = stmt.Exec()
if err != nil {
panic(err)
}
affected := stmt.GetAffectedRows()
fmt.Println("affected rows:", affected)
}
}
func prepare() {
db, err := sql.Open("taosRestful", "root:taosdata@http(localhost:6041)/")
if err != nil {
panic(err)
}
defer db.Close()
_, err = db.Exec("CREATE DATABASE IF NOT EXISTS power")
if err != nil {
panic(err)
}
_, err = db.Exec("CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))")
if err != nil {
panic(err)
}
}
无模式写入
- 原生连接
- WebSocket 连接
package main
import (
"fmt"
"github.com/taosdata/driver-go/v3/af"
)
const LineDemo = "meters,groupid=2,location=California.SanFrancisco current=10.3000002f64,voltage=219i32,phase=0.31f64 1626006833639000000"
const TelnetDemo = "stb0_0 1707095283260 4 host=host0 interface=eth0"
const JsonDemo = "{\"metric\": \"meter_current\",\"timestamp\": 1626846400,\"value\": 10.3, \"tags\": {\"groupid\": 2, \"location\": \"California.SanFrancisco\", \"id\": \"d1001\"}}"
func main() {
conn, err := af.Open("localhost", "root", "taosdata", "", 6030)
if err != nil {
fmt.Println("fail to connect, err:", err)
}
defer conn.Close()
_, err = conn.Exec("CREATE DATABASE IF NOT EXISTS power")
if err != nil {
panic(err)
}
_, err = conn.Exec("use power")
if err != nil {
panic(err)
}
err = conn.InfluxDBInsertLines([]string{LineDemo}, "ns")
if err != nil {
panic(err)
}
err = conn.OpenTSDBInsertTelnetLines([]string{TelnetDemo})
if err != nil {
panic(err)
}
err = conn.OpenTSDBInsertJsonPayload(JsonDemo)
if err != nil {
panic(err)
}
}
package main
import (
"database/sql"
"log"
"time"
"github.com/taosdata/driver-go/v3/common"
_ "github.com/taosdata/driver-go/v3/taosWS"
"github.com/taosdata/driver-go/v3/ws/schemaless"
)
const LineDemo = "meters,groupid=2,location=California.SanFrancisco current=10.3000002f64,voltage=219i32,phase=0.31f64 1626006833639000000"
const TelnetDemo = "stb0_0 1707095283260 4 host=host0 interface=eth0"
const JsonDemo = "{\"metric\": \"meter_current\",\"timestamp\": 1626846400,\"value\": 10.3, \"tags\": {\"groupid\": 2, \"location\": \"California.SanFrancisco\", \"id\": \"d1001\"}}"
func main() {
db, err := sql.Open("taosWS", "root:taosdata@ws(localhost:6041)/")
if err != nil {
log.Fatal(err)
}
defer db.Close()
_, err = db.Exec("create database if not exists power")
if err != nil {
log.Fatal(err)
}
s, err := schemaless.NewSchemaless(schemaless.NewConfig("ws://localhost:6041", 1,
schemaless.SetDb("power"),
schemaless.SetReadTimeout(10*time.Second),
schemaless.SetWriteTimeout(10*time.Second),
schemaless.SetUser("root"),
schemaless.SetPassword("taosdata"),
schemaless.SetErrorHandler(func(err error) {
log.Fatal(err)
}),
))
if err != nil {
panic(err)
}
err = s.Insert(LineDemo, schemaless.InfluxDBLineProtocol, "ns", 0, common.GetReqID())
if err != nil {
panic(err)
}
err = s.Insert(TelnetDemo, schemaless.OpenTSDBTelnetLineProtocol, "ms", 0, common.GetReqID())
if err != nil {
panic(err)
}
err = s.Insert(JsonDemo, schemaless.OpenTSDBJsonFormatProtocol, "ms", 0, common.GetReqID())
if err != nil {
panic(err)
}
}
执行带有 reqId 的无模式写入
func (s *Schemaless) Insert(lines string, protocol int, precision string, ttl int, reqID int64) error
可以通过 common.GetReqID()
获取唯一 id。
数据订阅
TDengine Go 连接器支持订阅功能,应用 API 如下:
创建 Topic
_, err = db.Exec("CREATE TOPIC IF NOT EXISTS topic_meters AS SELECT ts, current, voltage, phase, groupid, location FROM power.meters")
if err != nil {
panic(err)
}
创建 Consumer
consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{
"group.id": "test",
"auto.offset.reset": "latest",
"td.connect.ip": "127.0.0.1",
"td.connect.user": "root",
"td.connect.pass": "taosdata",
"td.connect.port": "6030",
"client.id": "test_tmq_client",
"enable.auto.commit": "false",
"msg.with.table.name": "true",
})
if err != nil {
panic(err)
}
订阅消费数据
go func() {
for {
_, err = db.Exec("insert into power.d001 values (now, 1.1, 220, 0.1)")
if err != nil {
panic(err)
}
time.Sleep(time.Millisecond * 100)
}
}()
err = consumer.Subscribe("topic_meters", nil)
if err != nil {
panic(err)
}
for i := 0; i < 5; i++ {
ev := consumer.Poll(500)
if ev != nil {
switch e := ev.(type) {
case *tmqcommon.DataMessage:
fmt.Printf("get message:%v\n", e)
case tmqcommon.Error:
fmt.Fprintf(os.Stderr, "%% Error: %v: %v\n", e.Code(), e)
panic(e)
}
consumer.Commit()
}
}
指定订阅 Offset
partitions, err := consumer.Assignment()
if err != nil {
panic(err)
}
for i := 0; i < len(partitions); i++ {
fmt.Println(partitions[i])
err = consumer.Seek(tmqcommon.TopicPartition{
Topic: partitions[i].Topic,
Partition: partitions[i].Partition,
Offset: 0,
}, 0)
if err != nil {
panic(err)
}
}
partitions, err = consumer.Assignment()
if err != nil {
panic(err)
}
关闭订阅
err = consumer.Close()
if err != nil {
panic(err)
}
完整示例
- 原生连接
- WebSocket 连接
package main
import (
"fmt"
"os"
"time"
"github.com/taosdata/driver-go/v3/af"
"github.com/taosdata/driver-go/v3/af/tmq"
tmqcommon "github.com/taosdata/driver-go/v3/common/tmq"
)
func main() {
db, err := af.Open("", "root", "taosdata", "", 0)
if err != nil {
panic(err)
}
defer db.Close()
_, err = db.Exec("create database if not exists power WAL_RETENTION_PERIOD 86400")
if err != nil {
panic(err)
}
_, err = db.Exec("CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))")
if err != nil {
panic(err)
}
_, err = db.Exec("create table if not exists power.d001 using power.meters tags(1,'location')")
if err != nil {
panic(err)
}
// ANCHOR: create_topic
_, err = db.Exec("CREATE TOPIC IF NOT EXISTS topic_meters AS SELECT ts, current, voltage, phase, groupid, location FROM power.meters")
if err != nil {
panic(err)
}
// ANCHOR_END: create_topic
// ANCHOR: create_consumer
consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{
"group.id": "test",
"auto.offset.reset": "latest",
"td.connect.ip": "127.0.0.1",
"td.connect.user": "root",
"td.connect.pass": "taosdata",
"td.connect.port": "6030",
"client.id": "test_tmq_client",
"enable.auto.commit": "false",
"msg.with.table.name": "true",
})
if err != nil {
panic(err)
}
// ANCHOR_END: create_consumer
// ANCHOR: poll_data
go func() {
for {
_, err = db.Exec("insert into power.d001 values (now, 1.1, 220, 0.1)")
if err != nil {
panic(err)
}
time.Sleep(time.Millisecond * 100)
}
}()
err = consumer.Subscribe("topic_meters", nil)
if err != nil {
panic(err)
}
for i := 0; i < 5; i++ {
ev := consumer.Poll(500)
if ev != nil {
switch e := ev.(type) {
case *tmqcommon.DataMessage:
fmt.Printf("get message:%v\n", e)
case tmqcommon.Error:
fmt.Fprintf(os.Stderr, "%% Error: %v: %v\n", e.Code(), e)
panic(e)
}
consumer.Commit()
}
}
// ANCHOR_END: poll_data
// ANCHOR: consumer_seek
partitions, err := consumer.Assignment()
if err != nil {
panic(err)
}
for i := 0; i < len(partitions); i++ {
fmt.Println(partitions[i])
err = consumer.Seek(tmqcommon.TopicPartition{
Topic: partitions[i].Topic,
Partition: partitions[i].Partition,
Offset: 0,
}, 0)
if err != nil {
panic(err)
}
}
partitions, err = consumer.Assignment()
if err != nil {
panic(err)
}
// ANCHOR_END: consumer_seek
for i := 0; i < len(partitions); i++ {
fmt.Println(partitions[i])
}
// ANCHOR: consumer_close
err = consumer.Close()
if err != nil {
panic(err)
}
// ANCHOR_END: consumer_close
}
package main
import (
"database/sql"
"fmt"
"os"
"time"
"github.com/taosdata/driver-go/v3/common"
tmqcommon "github.com/taosdata/driver-go/v3/common/tmq"
_ "github.com/taosdata/driver-go/v3/taosRestful"
"github.com/taosdata/driver-go/v3/ws/tmq"
)
func main() {
db, err := sql.Open("taosRestful", "root:taosdata@http(localhost:6041)/")
if err != nil {
panic(err)
}
defer db.Close()
_, err = db.Exec("create database if not exists power WAL_RETENTION_PERIOD 86400")
if err != nil {
panic(err)
}
_, err = db.Exec("CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))")
if err != nil {
panic(err)
}
_, err = db.Exec("create table if not exists power.d001 using power.meters tags(1,'location')")
if err != nil {
panic(err)
}
// ANCHOR: create_topic
_, err = db.Exec("CREATE TOPIC IF NOT EXISTS topic_meters AS SELECT ts, current, voltage, phase, groupid, location FROM power.meters")
if err != nil {
panic(err)
}
// ANCHOR_END: create_topic
// ANCHOR: create_consumer
consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{
"ws.url": "ws://127.0.0.1:6041",
"ws.message.channelLen": uint(0),
"ws.message.timeout": common.DefaultMessageTimeout,
"ws.message.writeWait": common.DefaultWriteWait,
"td.connect.user": "root",
"td.connect.pass": "taosdata",
"group.id": "example",
"client.id": "example_consumer",
"auto.offset.reset": "latest",
})
if err != nil {
panic(err)
}
// ANCHOR_END: create_consumer
// ANCHOR: poll_data
go func() {
for {
_, err = db.Exec("insert into power.d001 values (now, 1.1, 220, 0.1)")
if err != nil {
panic(err)
}
time.Sleep(time.Millisecond * 100)
}
}()
err = consumer.Subscribe("topic_meters", nil)
if err != nil {
panic(err)
}
for i := 0; i < 5; i++ {
ev := consumer.Poll(500)
if ev != nil {
switch e := ev.(type) {
case *tmqcommon.DataMessage:
fmt.Printf("get message:%v\n", e)
case tmqcommon.Error:
fmt.Fprintf(os.Stderr, "%% Error: %v: %v\n", e.Code(), e)
panic(e)
}
consumer.Commit()
}
}
// ANCHOR_END: poll_data
// ANCHOR: consumer_seek
partitions, err := consumer.Assignment()
if err != nil {
panic(err)
}
for i := 0; i < len(partitions); i++ {
fmt.Println(partitions[i])
err = consumer.Seek(tmqcommon.TopicPartition{
Topic: partitions[i].Topic,
Partition: partitions[i].Partition,
Offset: 0,
}, 0)
if err != nil {
panic(err)
}
}
partitions, err = consumer.Assignment()
if err != nil {
panic(err)
}
// ANCHOR_END: consumer_seek
for i := 0; i < len(partitions); i++ {
fmt.Println(partitions[i])
}
// ANCHOR: consumer_close
err = consumer.Close()
if err != nil {
panic(err)
}
// ANCHOR_END: consumer_close
}
更多示例程序
常见问题
-
database/sql 中 stmt(参数绑定)相关接口崩溃
REST 不支持参数绑定相关接口,建议使用
db.Exec
和db.Query
。 -
使用
use db
语句后执行其他语句报错[0x217] Database not specified or available
在 REST 接口中 SQL 语句的执行无上下文关联,使用
use db
语句不会生效,解决办法见上方使用限制章节。 -
使用 taosSql 不报错使用 taosRestful 报错
[0x217] Database not specified or available
因为 REST 接口无状态,使用
use db
语句不会生效,解决办法见上方使用限制章节。 -
readBufferSize
参数调大后无明显效果readBufferSize
调大后会减少获取结果时syscall
的调用。如果查询结果的数据量不大,修改该参数不会带来明显提升,如果该参数修改过大,瓶颈会在解析 JSON 数据。如果需要优化查询速度,需要根据实际情况调整该值来达到查询效果最优。 -
disableCompression
参数设置为false
时查询效率降低当
disableCompression
参数设置为false
时查询结果会使用gzip
压缩后传输,拿到数据后要先进行gzip
解压。 -
go get
命令无法获取包,或者获取包超时
设置 Go 代理 go env -w GOPROXY=https://goproxy.cn,direct
。
API 参考
全部 API 见 driver-go 文档