Golang SDK

新建实时同步任务

在 Dashboard 或者通过命令行工具创建任务，数据源类型使用 Auto Push (自主推送)。

引入 Golang SDK

需要在项目中导入 SDK 的包，进行 SDK 的使用。导入方式如下：

import "github.com/apache/inlong/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy"

数据上报流程

导入 SDK 后，通过实例化一个 Client 接口对象后，调用相关的同步（ Send() ）或 异步（ SendAsync() ）接口来完成单条数据的上报任务，SDK 内部会根据 StreamID 将相同 StreamID 的数据批量上报。发送 Demo 可参考 example_test.go。 整体流程包括以下三个步骤：

初始化 SDK

client, err := dataproxy.NewClient(
    dataproxy.WithGroupID("test"),              
dataproxy.WithURL("http://127.0.0.1:8083/inlong/manager/openapi/dataproxy/getIpList"),
    dataproxy.WithMetricsName("test"),
)
if err != nil {
    fmt.Println(err)
    return
}

参数说明：

• dataproxy.WithGroupID("test") 设置了 GroupID ；
• dataproxy.WithURL("http://127.0.0.1:8083/inlong/manager/openapi/dataproxy/getIpList") 设置了 Manager 的 URL ；
• dataproxy.WithMetricsName("test") 设置了这个 Client 的指标中标签 “name” 的值；

调用发送接口进行数据上报

SDK 的数据发送接口是协程安全的，支持以同步或者异步模式发送单条数据。Demo 里有同步的方式，也有异步的方式。

同步方式：

for i := 0; i < 1000; i++ {
    err := client.Send(context.Background(), dataproxy.Message{
        GroupID:  "test",
        StreamID: "test",
        Payload:  []byte("test|a|b|c"),
    })
    if err !=nil {
        fmt.Println(err)
    }
}

异步方式：

var success atomic.Uint64
var failed atomic.Uint64
for i := 0; i < 1000; i++ {
    client.SendAsync(context.Background(),
        dataproxy.Message{
            GroupID:  "test",
            StreamID: "test",
            Payload:  []byte("test|a|b|c"),
        },
        func(msg dataproxy.Message, err error) {
            if err != nil {
                success.Add(1)
            } else {
                failed.Add(1)
            }
        })
}
// wait async send finish
time.Sleep(3 * time.Second)
fmt.Println("success:", success.Load())
fmt.Println("failed:", failed.Load())

我们推荐使用异步的方式发送，因为同步方式是无并发的，调用发送请求后需要等到响应或者超时，才能发下一条，在需要高吞吐量的场景是无法满足需求的。

关闭 SDK

关闭 SDK 只需要简单的调用 Close() 方法：

client.Close()

注意事项

• GroupID 和 URL 是 SDK 初始化必选的配置，你可以使用 dataproxy.WithGroupID() 和 dataproxy.WithURL() 来设置这些配置；
• 当你在一个进程里初始化多个 Client 实例时，MetricsName 必须配置不同的名称，否则指标查询将会失败，你可以使用 dataproxy.WithMetricsName() 来设置它；
• SDK 的默认配置已经能够满足运营需求，除非你需要调大或者调小配置，一般不需要更改其他配置，配置的默认值请参考后面的章节；
• SDK 的同步发送方法 Send() 是无并发的，在发送一个消息后，需要等待响应或者等待到超时才会返回，非必要的情况下不建议使用同步方法；
• SDK 默认会对消息进行2次重传，如果依然失败，需要调用者来决定如何做下一步处理。

错误

常见错误如下：

配置项

配置项请参考 options.go

指标

指标请参考 metrics.go

错误码请参考 worker.go