安裝

• 下載原始碼包

go get github.com/aliyun/aliyun-tablestore-go-sdk/tunnel

• 安裝依賴

  • 可以在tunnel目錄下使用dep安裝依賴

    • 安裝dep
    • dep ensure -v
  • 也可以直接使用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,
})