Table Store实时数据通道服务Go SDK快速入门-阿里云开发者社区

开发者社区> 阿里云存储服务> 正文

Table Store实时数据通道服务Go SDK快速入门

简介: # Tunnel Service Go SDK ## 安装 * 下载源码包 ```bash go get github.com/aliyun/aliyun-tablestore-go-sdk/tunnel ``` * 安装依赖 * 可以在tunnel目录下使用dep安装依赖 * 安装[dep](https://github.

文档

安装

  • 下载源码包
go get github.com/aliyun/aliyun-tablestore-go-sdk/tunnel
  • 安装依赖

    1. 可以在tunnel目录下使用dep安装依赖
dep ensure -v
  1. 也可以直接使用go get安装依赖包:
go get -u go.uber.org/zap
go get github.com/cenkalti/backoff
go get github.com/golang/protobuf/proto
go get github.com/satori/go.uuid
go get github.com/stretchr/testify/assert
go get github.com/smartystreets/goconvey/convey
go get github.com/golang/mock/gomock
go get gopkg.in/natefinch/lumberjack.v2

快速开始

  • 初始化Tunnel client:
// endpoint是表格存储实例endpoint,如https://instance.cn-hangzhou.ots.aliyun.com
// instance为实例名称
// accessKeyId和accessKeySecret分别为访问表格存储服务的AccessKey的Id和Secret
tunnelClient := tunnel.NewTunnelClient(endpoint, instance,
   accessKeyId, accessKeySecret)
  • 创建新Tunnel:
req := &tunnel.CreateTunnelRequest{
   TableName:  "testTable",
   TunnelName: "testTunnel",
   Type:       tunnel.TunnelTypeBaseStream, //全量加增量类型Tunnel
}
resp, err := tunnelClient.CreateTunnel(req)
if err != nil {
   log.Fatal("create test tunnel failed", err)
}
log.Println("tunnel id is", resp.TunnelId)
  • 获取已有Tunnel信息:
req := &tunnel.DescribeTunnelRequest{
   TableName:  "testTable",
   TunnelName: "testTunnel",
}
resp, err := tunnelClient.DescribeTunnel(req)
if err != nil {
   log.Fatal("create test tunnel failed", err)
}
log.Println("tunnel id is", resp.Tunnel.TunnelId)
  • 注册callback,开始数据消费:
//用户定义消费callback函数
func exampleConsumeFunction(ctx *tunnel.ChannelContext, records []*tunnel.Record) error {
    fmt.Println("user-defined information", ctx.CustomValue)
    for _, rec := range records {
        fmt.Println("tunnel record detail:", rec.String())
    }
    fmt.Println("a round of records consumption finished")
    return nil
}
//配置callback到SimpleProcessFactory,配置消费端TunnelWorkerConfig
workConfig := &tunnel.TunnelWorkerConfig{
   ProcessorFactory: &tunnel.SimpleProcessFactory{
      CustomValue: "user custom interface{} value",
      ProcessFunc: exampleConsumeFunction,
   },
}
//使用TunnelDaemon持续消费指定tunnel
daemon := tunnel.NewTunnelDaemon(tunnelClient, tunnelId, workConfig)
log.Fatal(daemon.Run())
  • 删除Tunnel
req := &tunnel.DeleteTunnelRequest {
   TableName: "testTable",
   TunnelName: "testTunnel",
}
_, err := tunnelClient.DeleteTunnel(req)
if err != nil {
   log.Fatal("delete test tunnel failed", err)
}

配置项

  • tunnel client配置
    初始化tunnel client时可以通过NewTunnelClientWithConfig接口自定义客户端配置,使用不指定config初始化接口或者config为nil时会使用DefaultTunnelConfig:
var DefaultTunnelConfig = &TunnelConfig{
      //最大指数退避重试时间
      MaxRetryElapsedTime: 45 * time.Second,
      //HTTP请求超时时间
      RequestTimeout:      30 * time.Second,
      //http.DefaultTransport
      Transport:           http.DefaultTransport,
}
  • 数据消费worker配置
    TunnelWorkerConfig中包含了数据消费worker需要的配置,其中ProcessorFactory为必填项,其余字段不填将使用默认值,通常使用默认值即可:
type TunnelWorkerConfig struct {
   //worker同Tunnel服务的心跳超时时间,通常使用默认值即可
   HeartbeatTimeout  time.Duration
   //worker发送心跳的频率,通常使用默认值即可
   HeartbeatInterval time.Duration
   //tunnel下消费连接建立接口,通常使用默认值即可
   ChannelDialer     ChannelDialer

   //消费连接上具体处理器产生接口,通常使用callback函数初始化SimpleProcessFactory即可
   ProcessorFactory ChannelProcessorFactory

   //zap日志配置,默认值为DefaultLogConfig
   LogConfig      *zap.Config
   //zap日志轮转配置,默认值为DefaultSyncer
   LogWriteSyncer zapcore.WriteSyncer
}

其中的ProcessorFactory为用户注册消费callback函数以及其他信息的接口,建议使用SDK中自带SimpleProcessorFactory实现:

type SimpleProcessFactory struct {
   //用户自定义信息,会传递到ProcessFunc和ShutdownFunc中的ChannelContext参数中
   CustomValue interface{}

   //Worker记录checkpoint的间隔,CpInterval<=0时会使用DefaultCheckpointInterval
   CpInterval time.Duration

   //worker数据处理的同步调用callback,ProcessFunc返回error时worker会用本批数据退避重试ProcessFunc
   ProcessFunc  func(channelCtx *ChannelContext, records []*Record) error
   //worker退出时的同步调用callback
   ShutdownFunc func(channelCtx *ChannelContext)

   //日志配置,Logger为nil时会使用DefaultLogConfig初始化logger
   Logger *zap.Logger
}
  • 日志配置
    默认日志配置:
//DefaultLogConfig是TunnelWorkerConfig和SimpleProcessFactory使用的默认日志配置
var DefaultLogConfig = zap.Config{
   Level:       zap.NewAtomicLevelAt(zap.InfoLevel),
   Development: false,
   Sampling: &zap.SamplingConfig{
      Initial:    100,
      Thereafter: 100,
   },
   Encoding: "json",
   EncoderConfig: zapcore.EncoderConfig{
      TimeKey:        "ts",
      LevelKey:       "level",
      NameKey:        "logger",
      CallerKey:      "caller",
      MessageKey:     "msg",
      StacktraceKey:  "stacktrace",
      LineEnding:     zapcore.DefaultLineEnding,
      EncodeLevel:    zapcore.LowercaseLevelEncoder,
      EncodeTime:     zapcore.ISO8601TimeEncoder,
      EncodeDuration: zapcore.SecondsDurationEncoder,
      EncodeCaller:   zapcore.ShortCallerEncoder,
   },
}

日志轮转配置:

//DefaultSyncer是TunnelWorkerConfig和SimpleProcessFactory使用的默认日志轮转配置
var DefaultSyncer = zapcore.AddSync(&lumberjack.Logger{
   //日志文件路径
   Filename:   "tunnelClient.log",
   //最大日志文件大小
   MaxSize:    512, //MB
   //压缩轮转的日志文件数
   MaxBackups: 5,
   //轮转日志文件保留的最大天数
   MaxAge:     30, //days
   //是否压缩轮转日志文件
   Compress:   true,
})

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:

阿里云存储基于飞天盘古2.0分布式存储系统,产品多种多样,充分满足用户数据存储和迁移上云需求。

官方博客
链接