05-go.mdx

---
title: TDengine Go Connector
sidebar_label: Go
description: This document describes the TDengine Go connector.
toc_max_heading_level: 4
---

import Tabs from '@theme/Tabs';
import TabItem from '@theme/TabItem';

import GoInsert from "../../07-develop/03-insert-data/_go_sql.mdx"
import GoInfluxLine from "../../07-develop/03-insert-data/_go_line.mdx"
import GoOpenTSDBTelnet from "../../07-develop/03-insert-data/_go_opts_telnet.mdx"
import GoOpenTSDBJson from "../../07-develop/03-insert-data/_go_opts_json.mdx"
import GoQuery from "../../07-develop/04-query-data/_go.mdx"

`driver-go` is the official Go language connector for TDengine. It implements the [database/sql](https://golang.org/pkg/database/sql/) package, the generic Go language interface to SQL databases. Go developers can use it to develop applications that access TDengine cluster data.

`driver-go` provides two ways to establish connections. One is **native connection**, which connects to TDengine instances natively through the TDengine client driver (taosc), supporting data writing, querying, subscriptions, schemaless writing, and bind interface. The other is the **REST connection**, which connects to TDengine instances via the REST interface provided by taosAdapter. The set of features implemented by the REST connection differs slightly from those implemented by the native connection.

This article describes how to install `driver-go` and connect to TDengine clusters and perform basic operations such as data query and data writing through `driver-go`.

The source code of `driver-go` is hosted on [GitHub](https://github.com/taosdata/driver-go).

## Supported platforms

Native connections are supported on the same platforms as the TDengine client driver.
REST connections are supported on all platforms that can run Go.

## Version support

Please refer to [version support list](https://github.com/taosdata/driver-go#remind)

## Handling exceptions

If it is a TDengine error, you can get the error code and error information in the following ways.
```go
// 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 vs. 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    |

**Note**: Only TAG supports JSON types

## Installation Steps

### Pre-installation preparation

* Install Go development environment (Go 1.14 and above, GCC 4.8.5 and above)
* If you use the native connector, please install the TDengine client driver. Please refer to [Install Client Driver](/reference/connector/#install-client-driver) for specific steps

Configure the environment variables and check the command.

* ```go env```
* ```gcc -v```

### Install the connectors

1. Initialize the project with the `go mod` command.

   ```text
   go mod init taos-demo
   ```

2. Introduce taosSql

   ```go
   import (
     "database/sql"
     _ "github.com/taosdata/driver-go/v3/taosSql"
   )
   ```

3. Update the dependency packages with `go mod tidy`.

   ```text
   go mod tidy
   ```

4. Run the program with `go run taos-demo` or compile the binary with the `go build` command.

  ```text
  go run taos-demo
  go build
  ```

## Establishing a connection

Data source names have a standard format, e.g. [PEAR DB](http://pear.php.net/manual/en/package.database.db.intro-dsn.php), but no type prefix (square brackets indicate optionally):

``` text
[username[:password]@][protocol[(address)]]/[dbname][?param1=value1&...&paramN=valueN]
```

DSN in full form.

```text
username:password@protocol(address)/dbname?param=value
```
<Tabs defaultValue="rest" groupId="connect">
<TabItem value="native" label="native connection">

_taosSql_ implements Go's `database/sql/driver` interface via cgo. You can use the [`database/sql`](https://golang.org/pkg/database/sql/) interface by simply introducing the driver.

Use `taosSql` as `driverName` and use a correct [DSN](#DSN) as `dataSourceName`, DSN supports the following parameters.

* cfg specifies the `taos.cfg` directory

For example:

```go
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
    }
}
```

</TabItem>
<TabItem value="rest" label="REST connection">

_taosRestful_ implements Go's `database/sql/driver` interface via `http client`. You can use the [`database/sql`](https://golang.org/pkg/database/sql/) interface by simply introducing the driver.

Use `taosRestful` as `driverName` and use a correct [DSN](#DSN) as `dataSourceName` with the following parameters supported by the DSN.

* `disableCompression` whether to accept compressed data, default is true do not accept compressed data, set to false if transferring data using gzip compression.
* `readBufferSize` The default size of the buffer for reading data is 4K (4096), which can be adjusted upwards when the query result has a lot of data.

For example:

```go
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
    }
}
```
</TabItem>
<TabItem value="WebSocket" label="WebSocket connection">

_taosRestful_ implements Go's `database/sql/driver` interface via `http client`. You can use the [`database/sql`](https://golang.org/pkg/database/sql/) interface by simply introducing the driver (driver-go minimum version 3.0.2).

Use `taosWS` as `driverName` and use a correct [DSN](#DSN) as `dataSourceName` with the following parameters supported by the DSN.

* `writeTimeout` The timeout to send data via WebSocket.
* `readTimeout` The timeout to receive response data via WebSocket.

For example:

```go
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
    }
}
```
</TabItem>
</Tabs>

### Specify the URL and Properties to get the connection

The Go connector does not support this feature

### Priority of configuration parameters

The Go connector does not support this feature

## Usage examples

### Create database and tables

```go
var taosDSN = "root:taosdata@tcp(localhost:6030)/"
taos, err := sql.Open("taosSql", taosDSN)
if err != nil {
    log.Fatalln("failed to connect TDengine, err:", err)
}
defer taos.Close()
_, err := taos.Exec("CREATE DATABASE power")
if err != nil {
    log.Fatalln("failed to create database, err:", err)
}
_, err = taos.Exec("CREATE STABLE power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (location BINARY(64), groupId INT)")
if err != nil {
    log.Fatalln("failed to create stable, err:", err)
}
```

### Insert data

<GoInsert />

### Querying data

<GoQuery />

### execute SQL with reqId

This reqId can be used to request link tracing.

```go
db, err := sql.Open("taosSql", "root:taosdata@tcp(localhost:6030)/")
if err != nil {
    panic(err)
}
defer db.Close()
ctx := context.WithValue(context.Background(), common.ReqIDKey, common.GetReqID())
_, err = db.ExecContext(ctx, "create database if not exists example_taos_sql")
if err != nil {
    panic(err)
}
```

### Writing data via parameter binding

<Tabs defaultValue="native" groupId="connect">
<TabItem value="native" label="native connection">

```go
package main

import (
    "time"

    "github.com/taosdata/driver-go/v3/af"
    "github.com/taosdata/driver-go/v3/common"
    "github.com/taosdata/driver-go/v3/common/param"
)

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 example_stmt")
    if err != nil {
        panic(err)
    }
    _, err = db.Exec("create table if not exists example_stmt.tb1(ts timestamp," +
        "c1 bool," +
        "c2 tinyint," +
        "c3 smallint," +
        "c4 int," +
        "c5 bigint," +
        "c6 tinyint unsigned," +
        "c7 smallint unsigned," +
        "c8 int unsigned," +
        "c9 bigint unsigned," +
        "c10 float," +
        "c11 double," +
        "c12 binary(20)," +
        "c13 nchar(20)" +
        ")")
    if err != nil {
        panic(err)
    }
    stmt := db.InsertStmt()
    err = stmt.Prepare("insert into example_stmt.tb1 values(?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
    if err != nil {
        panic(err)
    }
    now := time.Now()
    params := make([]*param.Param, 14)
    params[0] = param.NewParam(2).
        AddTimestamp(now, common.PrecisionMilliSecond).
        AddTimestamp(now.Add(time.Second), common.PrecisionMilliSecond)
    params[1] = param.NewParam(2).AddBool(true).AddNull()
    params[2] = param.NewParam(2).AddTinyint(2).AddNull()
    params[3] = param.NewParam(2).AddSmallint(3).AddNull()
    params[4] = param.NewParam(2).AddInt(4).AddNull()
    params[5] = param.NewParam(2).AddBigint(5).AddNull()
    params[6] = param.NewParam(2).AddUTinyint(6).AddNull()
    params[7] = param.NewParam(2).AddUSmallint(7).AddNull()
    params[8] = param.NewParam(2).AddUInt(8).AddNull()
    params[9] = param.NewParam(2).AddUBigint(9).AddNull()
    params[10] = param.NewParam(2).AddFloat(10).AddNull()
    params[11] = param.NewParam(2).AddDouble(11).AddNull()
    params[12] = param.NewParam(2).AddBinary([]byte("binary")).AddNull()
    params[13] = param.NewParam(2).AddNchar("nchar").AddNull()

    paramTypes := param.NewColumnType(14).
        AddTimestamp().
        AddBool().
        AddTinyint().
        AddSmallint().
        AddInt().
        AddBigint().
        AddUTinyint().
        AddUSmallint().
        AddUInt().
        AddUBigint().
        AddFloat().
        AddDouble().
        AddBinary(6).
        AddNchar(5)
    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)
    }
    err = stmt.Close()
    if err != nil {
        panic(err)
    }
    // select * from example_stmt.tb1
}
```

</TabItem>
<TabItem value="WebSocket" label="WebSocket connection">

```go
package main

import (
    "database/sql"
    "fmt"
    "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"
)

func main() {
    db, err := sql.Open("taosRestful", "root:taosdata@http(localhost:6041)/")
    if err != nil {
        panic(err)
    }
    defer db.Close()
    prepareEnv(db)

    config := stmt.NewConfig("ws://127.0.0.1:6041/rest/stmt", 0)
    config.SetConnectUser("root")
    config.SetConnectPass("taosdata")
    config.SetConnectDB("example_ws_stmt")
    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)
    }
    now := time.Now()
    {
        stmt, err := connector.Init()
        if err != nil {
            panic(err)
        }
        err = stmt.Prepare("insert into ? using all_json tags(?) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
        if err != nil {
            panic(err)
        }
        err = stmt.SetTableName("tb1")
        if err != nil {
            panic(err)
        }
        err = stmt.SetTags(param.NewParam(1).AddJson([]byte(`{"tb":1}`)), param.NewColumnType(1).AddJson(0))
        if err != nil {
            panic(err)
        }
        params := []*param.Param{
            param.NewParam(3).AddTimestamp(now, 0).AddTimestamp(now.Add(time.Second), 0).AddTimestamp(now.Add(time.Second*2), 0),
            param.NewParam(3).AddBool(true).AddNull().AddBool(true),
            param.NewParam(3).AddTinyint(1).AddNull().AddTinyint(1),
            param.NewParam(3).AddSmallint(1).AddNull().AddSmallint(1),
            param.NewParam(3).AddInt(1).AddNull().AddInt(1),
            param.NewParam(3).AddBigint(1).AddNull().AddBigint(1),
            param.NewParam(3).AddUTinyint(1).AddNull().AddUTinyint(1),
            param.NewParam(3).AddUSmallint(1).AddNull().AddUSmallint(1),
            param.NewParam(3).AddUInt(1).AddNull().AddUInt(1),
            param.NewParam(3).AddUBigint(1).AddNull().AddUBigint(1),
            param.NewParam(3).AddFloat(1).AddNull().AddFloat(1),
            param.NewParam(3).AddDouble(1).AddNull().AddDouble(1),
            param.NewParam(3).AddBinary([]byte("test_binary")).AddNull().AddBinary([]byte("test_binary")),
            param.NewParam(3).AddNchar("test_nchar").AddNull().AddNchar("test_nchar"),
        }
        paramTypes := param.NewColumnType(14).
            AddTimestamp().
            AddBool().
            AddTinyint().
            AddSmallint().
            AddInt().
            AddBigint().
            AddUTinyint().
            AddUSmallint().
            AddUInt().
            AddUBigint().
            AddFloat().
            AddDouble().
            AddBinary(0).
            AddNchar(0)
        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("all_json affected rows:", affected)
        err = stmt.Close()
        if err != nil {
            panic(err)
        }
    }
    {
        stmt, err := connector.Init()
        if err != nil {
            panic(err)
        }
        err = stmt.Prepare("insert into ? using all_all tags(?,?,?,?,?,?,?,?,?,?,?,?,?,?) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
        err = stmt.SetTableName("tb1")
        if err != nil {
            panic(err)
        }

        err = stmt.SetTableName("tb2")
        if err != nil {
            panic(err)
        }
        err = stmt.SetTags(
            param.NewParam(14).
                AddTimestamp(now, 0).
                AddBool(true).
                AddTinyint(2).
                AddSmallint(2).
                AddInt(2).
                AddBigint(2).
                AddUTinyint(2).
                AddUSmallint(2).
                AddUInt(2).
                AddUBigint(2).
                AddFloat(2).
                AddDouble(2).
                AddBinary([]byte("tb2")).
                AddNchar("tb2"),
            param.NewColumnType(14).
                AddTimestamp().
                AddBool().
                AddTinyint().
                AddSmallint().
                AddInt().
                AddBigint().
                AddUTinyint().
                AddUSmallint().
                AddUInt().
                AddUBigint().
                AddFloat().
                AddDouble().
                AddBinary(0).
                AddNchar(0),
        )
        if err != nil {
            panic(err)
        }
        params := []*param.Param{
            param.NewParam(3).AddTimestamp(now, 0).AddTimestamp(now.Add(time.Second), 0).AddTimestamp(now.Add(time.Second*2), 0),
            param.NewParam(3).AddBool(true).AddNull().AddBool(true),
            param.NewParam(3).AddTinyint(1).AddNull().AddTinyint(1),
            param.NewParam(3).AddSmallint(1).AddNull().AddSmallint(1),
            param.NewParam(3).AddInt(1).AddNull().AddInt(1),
            param.NewParam(3).AddBigint(1).AddNull().AddBigint(1),
            param.NewParam(3).AddUTinyint(1).AddNull().AddUTinyint(1),
            param.NewParam(3).AddUSmallint(1).AddNull().AddUSmallint(1),
            param.NewParam(3).AddUInt(1).AddNull().AddUInt(1),
            param.NewParam(3).AddUBigint(1).AddNull().AddUBigint(1),
            param.NewParam(3).AddFloat(1).AddNull().AddFloat(1),
            param.NewParam(3).AddDouble(1).AddNull().AddDouble(1),
            param.NewParam(3).AddBinary([]byte("test_binary")).AddNull().AddBinary([]byte("test_binary")),
            param.NewParam(3).AddNchar("test_nchar").AddNull().AddNchar("test_nchar"),
        }
        paramTypes := param.NewColumnType(14).
            AddTimestamp().
            AddBool().
            AddTinyint().
            AddSmallint().
            AddInt().
            AddBigint().
            AddUTinyint().
            AddUSmallint().
            AddUInt().
            AddUBigint().
            AddFloat().
            AddDouble().
            AddBinary(0).
            AddNchar(0)
        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("all_all affected rows:", affected)
        err = stmt.Close()
        if err != nil {
            panic(err)
        }

    }
}

func prepareEnv(db *sql.DB) {
    steps := []string{
        "create database example_ws_stmt",
        "create table example_ws_stmt.all_json(ts timestamp," +
            "c1 bool," +
            "c2 tinyint," +
            "c3 smallint," +
            "c4 int," +
            "c5 bigint," +
            "c6 tinyint unsigned," +
            "c7 smallint unsigned," +
            "c8 int unsigned," +
            "c9 bigint unsigned," +
            "c10 float," +
            "c11 double," +
            "c12 binary(20)," +
            "c13 nchar(20)" +
            ")" +
            "tags(t json)",
        "create table example_ws_stmt.all_all(" +
            "ts timestamp," +
            "c1 bool," +
            "c2 tinyint," +
            "c3 smallint," +
            "c4 int," +
            "c5 bigint," +
            "c6 tinyint unsigned," +
            "c7 smallint unsigned," +
            "c8 int unsigned," +
            "c9 bigint unsigned," +
            "c10 float," +
            "c11 double," +
            "c12 binary(20)," +
            "c13 nchar(20)" +
            ")" +
            "tags(" +
            "tts timestamp," +
            "tc1 bool," +
            "tc2 tinyint," +
            "tc3 smallint," +
            "tc4 int," +
            "tc5 bigint," +
            "tc6 tinyint unsigned," +
            "tc7 smallint unsigned," +
            "tc8 int unsigned," +
            "tc9 bigint unsigned," +
            "tc10 float," +
            "tc11 double," +
            "tc12 binary(20)," +
            "tc13 nchar(20))",
    }
    for _, step := range steps {
        _, err := db.Exec(step)
        if err != nil {
            panic(err)
        }
    }
}

```

</TabItem>
</Tabs>


### Schemaless Writing

<Tabs defaultValue="native" groupId="connect">
<TabItem value="native" label="native connection">

```go
import (
    "fmt"

    "github.com/taosdata/driver-go/v3/af"
)

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 example")
    if err != nil {
        panic(err)
    }
    _, err = conn.Exec("use example")
    if err != nil {
        panic(err)
    }
    influxdbData := "st,t1=3i64,t2=4f64,t3=\"t3\" c1=3i64,c3=L\"passit\",c2=false,c4=4f64 1626006833639000000"
    err = conn.InfluxDBInsertLines([]string{influxdbData}, "ns")
    if err != nil {
        panic(err)
    }
    telnetData := "stb0_0 1626006833 4 host=host0 interface=eth0"
    err = conn.OpenTSDBInsertTelnetLines([]string{telnetData})
    if err != nil {
        panic(err)
    }
    jsonData := "{\"metric\": \"meter_current\",\"timestamp\": 1626846400,\"value\": 10.3, \"tags\": {\"groupid\": 2, \"location\": \"California.SanFrancisco\", \"id\": \"d1001\"}}"
    err = conn.OpenTSDBInsertJsonPayload(jsonData)
    if err != nil {
        panic(err)
    }
}    
```

</TabItem>
<TabItem value="WebSocket" label="WebSocket connection">

```go
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"
)

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 schemaless_ws")
    if err != nil {
        log.Fatal(err)
    }
    s, err := schemaless.NewSchemaless(schemaless.NewConfig("ws://localhost:6041/rest/schemaless", 1,
        schemaless.SetDb("schemaless_ws"),
        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)
    }
    influxdbData := "st,t1=3i64,t2=4f64,t3=\"t3\" c1=3i64,c3=L\"passit\",c2=false,c4=4f64 1626006833639000000"
    telnetData := "stb0_0 1626006833 4 host=host0 interface=eth0"
    jsonData := "{\"metric\": \"meter_current\",\"timestamp\": 1626846400,\"value\": 10.3, \"tags\": {\"groupid\": 2, \"location\": \"California.SanFrancisco\", \"id\": \"d1001\"}}"

    err = s.Insert(influxdbData, schemaless.InfluxDBLineProtocol, "ns", 0, common.GetReqID())
    if err != nil {
        panic(err)
    }
    err = s.Insert(telnetData, schemaless.OpenTSDBTelnetLineProtocol, "ms", 0, common.GetReqID())
    if err != nil {
        panic(err)
    }
    err = s.Insert(jsonData, schemaless.OpenTSDBJsonFormatProtocol, "ms", 0, common.GetReqID())
    if err != nil {
        panic(err)
    }
}
```

</TabItem>
</Tabs>


### Schemaless with reqId

```go
func (s *Schemaless) Insert(lines string, protocol int, precision string, ttl int, reqID int64) error
```

You can get the unique id by `common.GetReqID()`.

### Data Subscription

The TDengine Go Connector supports subscription functionality with the following application API.

#### Create a Topic

```go
    db, err := af.Open("", "root", "taosdata", "", 0)
    if err != nil {
        panic(err)
    }
    defer db.Close()
    _, err = db.Exec("create database if not exists example_tmq WAL_RETENTION_PERIOD 86400")
    if err != nil {
        panic(err)
    }
    _, err = db.Exec("create topic if not exists example_tmq_topic as DATABASE example_tmq")
    if err != nil {
        panic(err)
    }
```

#### Create a Consumer

```go
    consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{
        "group.id":                     "test",
        "auto.offset.reset":            "earliest",
        "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)
    }
```

#### Subscribe to consume data

```go
    err = consumer.Subscribe("example_tmq_topic", 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()
        }
    }
```

#### Assignment subscription Offset

```go
    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)
        }
    }
```

#### Close subscriptions

```go
    err = consumer.Close()
    if err != nil {
        panic(err)
    }
```

#### Full Sample Code

<Tabs defaultValue="native" groupId="connect">
<TabItem value="native" label="native connection">

```go
package main

import (
    "fmt"
    "os"

    "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 example_tmq WAL_RETENTION_PERIOD 86400")
    if err != nil {
        panic(err)
    }
    _, err = db.Exec("create topic if not exists example_tmq_topic as DATABASE example_tmq")
    if err != nil {
        panic(err)
    }
    if err != nil {
        panic(err)
    }
    consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{
        "group.id":                     "test",
        "auto.offset.reset":            "earliest",
        "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)
    }
    err = consumer.Subscribe("example_tmq_topic", nil)
    if err != nil {
        panic(err)
    }
    _, err = db.Exec("create table example_tmq.t1 (ts timestamp,v int)")
    if err != nil {
        panic(err)
    }
    _, err = db.Exec("insert into example_tmq.t1 values(now,1)")
    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()
        }
    }
    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)
    }
    for i := 0; i < len(partitions); i++ {
        fmt.Println(partitions[i])
    }

    err = consumer.Close()
    if err != nil {
        panic(err)
    }
}
```

</TabItem>
<TabItem value="WebSocket" label="WebSocket connection">

```go
package main

import (
    "database/sql"
    "fmt"

    "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()
    prepareEnv(db)
    consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{
        "ws.url":                "ws://127.0.0.1:6041/rest/tmq",
        "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":     "earliest",
    })
    if err != nil {
        panic(err)
    }
    err = consumer.Subscribe("example_ws_tmq_topic", nil)
    if err != nil {
        panic(err)
    }
    go func() {
        _, err := db.Exec("create table example_ws_tmq.t_all(ts timestamp," +
            "c1 bool," +
            "c2 tinyint," +
            "c3 smallint," +
            "c4 int," +
            "c5 bigint," +
            "c6 tinyint unsigned," +
            "c7 smallint unsigned," +
            "c8 int unsigned," +
            "c9 bigint unsigned," +
            "c10 float," +
            "c11 double," +
            "c12 binary(20)," +
            "c13 nchar(20)" +
            ")")
        if err != nil {
            panic(err)
        }
        _, err = db.Exec("insert into example_ws_tmq.t_all values(now,true,2,3,4,5,6,7,8,9,10.123,11.123,'binary','nchar')")
        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.Printf("%% Error: %v: %v\n", e.Code(), e)
                panic(e)
            }
            consumer.Commit()
        }
    }
    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)
    }
    for i := 0; i < len(partitions); i++ {
        fmt.Println(partitions[i])
    }

    err = consumer.Close()
    if err != nil {
        panic(err)
    }
}

func prepareEnv(db *sql.DB) {
    _, err := db.Exec("create database example_ws_tmq WAL_RETENTION_PERIOD 86400")
    if err != nil {
        panic(err)
    }
    _, err = db.Exec("create topic example_ws_tmq_topic as database example_ws_tmq")
    if err != nil {
        panic(err)
    }
}
```

</TabItem>
</Tabs>

### More sample programs

* [sample program](https://github.com/taosdata/driver-go/tree/3.0/examples)


## Frequently Asked Questions

1. bind interface in database/sql crashes

   REST does not support parameter binding related interface. It is recommended to use `db.Exec` and `db.Query`.

2. error `[0x217] Database not specified or available` after executing other statements with `use db` statement

   The execution of SQL command in the REST interface is not contextual, so using `use db` statement will not work, see the usage restrictions section above.

3. use `taosSql` without error but use `taosRestful` with error `[0x217] Database not specified or available`

   Because the REST interface is stateless, using the `use db` statement will not take effect. See the usage restrictions section above.

4. `readBufferSize` parameter has no significant effect after being increased

   Increasing `readBufferSize` will reduce the number of `syscall` calls when fetching results. If the query result is smaller, modifying this parameter will not improve performance significantly. If you increase the parameter value too much, the bottleneck will be parsing JSON data. If you need to optimize the query speed, you must adjust the value based on the actual situation to achieve the best query performance.

5. `disableCompression` parameter is set to `false` when the query efficiency is reduced

   When set `disableCompression` parameter to `false`, the query result will be compressed by `gzip` and then transmitted, so you have to decompress the data by `gzip` after getting it.

6. `go get` command can't get the package, or timeout to get the package

   Set Go proxy `go env -w GOPROXY=https://goproxy.cn,direct`.

## API Reference

Full API see [driver-go documentation](https://pkg.go.dev/github.com/taosdata/driver-go/v3)