Tablestore:Giriş

Kullanım notları

• Sistem varsayılan olarak veri okumak ve işlemek için TunnelWorkerConfig'a dayalı bir iş parçacığı havuzu başlatır. Tek bir sunucuda birden fazla TunnelWorker çalıştırmak istiyorsanız tüm TunnelWorker'ları aynı TunnelWorkerConfig ile yapılandırmanızı öneririz.

• TunnelWorker'ın başlatma aşamasında bir ısınma süresine ihtiyacı vardır. Bu süre TunnelWorkerConfig içindeki HeartbeatInterval parametresiyle belirtilir. Varsayılan değer 30 saniyedir.

• TunnelWorker istemcisi beklenmeyen bir çıkış veya manuel sonlandırma nedeniyle kapandığında TunnelWorker kaynakları otomatik olarak geri kazanır. Bunun için iş parçacığı havuzunu serbest bırakır, Channel sınıfı için kaydettiğiniz shutdown yöntemini çağırır ve tüneli kapatır.

• Tünellerdeki artımlı günlüklerin saklama süresi Stream günlüklerinin saklama süresiyle aynıdır. Stream günlükleri en fazla yedi gün saklanabilir. Dolayısıyla tünellerdeki artımlı günlükler de en fazla yedi gün saklanabilir.

• Farklı veya artımlı verileri tüketmek üzere bir tünel oluştururken aşağıdaki hususlara dikkat edin:

  • Tam veri tüketimi sırasında tünel artımlı günlüklerin saklama süresi (en fazla yedi gün) içinde tam veri tüketimini tamamlayamazsa tünel artımlı günlükleri tüketmeye başladığında OTSTunnelExpired hatası oluşur. Bu durumda tünel artımlı günlükleri tüketemez.

    Tünelin belirtilen zaman aralığında tam veri tüketimini tamamlayamayacağını tahmin ediyorsanız Tablestore teknik desteğiyle iletişime geçin.

  • Artımlı veri tüketimi sırasında tünel artımlı günlüklerin saklama süresi (en fazla yedi gün) içinde artımlı günlüklerin tüketimini tamamlayamazsa tünel verileri en yeni mevcut veriden itibaren tüketmeye başlayabilir. Bu durumda belirli veriler tüketilmeyebilir.

• Bir tünelin süresi dolduktan sonra Tablestore tüneli devre dışı bırakabilir. Bir tünel 30 günden uzun süre devre dışı durumda kalırsa silinir. Silinen bir tüneli geri yükleyemezsiniz.

Ön koşullar

• Bir veri tablosu oluşturun. Daha fazla bilgi için bkz. Tablestore konsolunu kullanma, Tablestore CLI'yı kullanma ve Tablestore SDK'larını kullanma.

• Veri tablosunun bulunduğu örneğin uç noktasını edinin. Daha fazla bilgi için bkz. Tablestore örneği için uç nokta edinme.

• Erişim kimlik bilgilerini yapılandırın. Daha fazla bilgi için bkz. Erişim kimlik bilgilerini yapılandırma.

Tunnel Service kullanmaya başlama

1. Bir TunnelClient örneğini başlatın.

   Bir TunnelClient örneğini başlatırken kimlik doğrulama için uzun süreli erişim kimlik bilgilerini veya geçici erişim kimlik bilgilerini kullanabilirsiniz.

   • Başlatma için uzun süreli erişim kimlik bilgilerini kullanma

     TABLESTORE_ACCESS_KEY_ID ve TABLESTORE_ACCESS_KEY_SECRET ortam değişkenlerinin yapılandırıldığından emin olun. TABLESTORE_ACCESS_KEY_ID ortam değişkeni Alibaba Cloud hesabınızın veya RAM kullanıcınızın AccessKey ID'sini belirtir. TABLESTORE_ACCESS_KEY_SECRET ortam değişkeni Alibaba Cloud hesabınızın veya RAM kullanıcınızın AccessKey secret'ını belirtir.

     Uyarı

     Alibaba Cloud hesabı hesaba ait tüm kaynaklara tam erişime sahiptir. Alibaba Cloud hesabının AccessKey çiftinin sızması sisteme yönelik kritik tehditler oluşturur. Bu nedenle TunnelClient örneğini başlatmak için minimum gerekli izinlere sahip bir RAM kullanıcısının AccessKey çiftini kullanmanızı öneririz.

     // Set the endpoint parameter to the endpoint of the Tablestore instance. Example: https://instance.cn-hangzhou.ots.aliyuncs.com. 
     // Specify the name of the instance. 
     // Specify the AccessKey ID and AccessKey secret of your Alibaba Cloud account or a RAM user. 
     endpoint := "yourEndpoint"
     instance := "yourInstance"
     accessKeyId := os.Getenv("TABLESTORE_ACCESS_KEY_ID")
     accessKeySecret := os.Getenv("TABLESTORE_ACCESS_KEY_SECRET")
     tunnelClient := tunnel.NewTunnelClient(endpoint, instance, accessKeyId, accessKeySecret)                    

   • Başlatma için geçici erişim kimlik bilgilerini kullanma

     1. Tablestore'a geçici olarak erişmek için Go için Tablestore SDK'yı kullanmak istiyorsanız geçici erişim kimlik bilgileri oluşturmak üzere Security Token Service (STS) kullanabilirsiniz. Daha fazla bilgi için bkz. Geçici erişim kimlik bilgilerini yapılandırma.

     2. Bir tünel istemcisi geçici erişim kimlik bilgilerine dayalı olarak bir TunnelClient örneğini başlatmak için çağırabileceğiniz NewTunnelClientWithToken işlemini sağlar. Bu konuda periyodik olarak yenilenebilen geçici erişim kimlik bilgilerini kullanarak bir TunnelClient örneğinin başlatılmasına ilişkin örnek kod sağlanmaktadır. Daha fazla bilgi için bkz. Ek: Geçici erişim kimlik bilgilerini kullanarak bir TunnelClient örneğini başlatmaya ilişkin örnek kod.

2. Bir tünel oluşturun.

   req := &tunnel.CreateTunnelRequest{
      TableName:  "<TABLE_NAME>",
      TunnelName: "<TUNNEL_NAME>",
      Type:       tunnel.TunnelTypeBaseStream, // Create a BaseAndStream tunnel. 
   }
   resp, err := tunnelClient.CreateTunnel(req)
   if err != nil {
      log.Fatal("create test tunnel failed", err)
   }
   log.Println("tunnel id is", resp.TunnelId)

3. Otomatik veri tüketimini başlatmak için özel bir geri arama işlevi belirtin.

   // Specify a custom callback function. 
   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
   }
   
   // Configure the callback function. Information about the callback function is passed to SimpleProcessFactory. Configure TunnelWorkerConfig for the consumer. 
   workConfig := &tunnel.TunnelWorkerConfig{
      ProcessorFactory: &tunnel.SimpleProcessFactory{
         CustomValue: "user custom interface{} value",
         ProcessFunc: exampleConsumeFunction,
      },
   }
   
   // Use TunnelDaemon to continuously consume the specified tunnel. 
   tunnelId := "<TUNNEL_ID>"
   daemon := tunnel.NewTunnelDaemon(tunnelClient, tunnelId, workConfig)
   log.Fatal(daemon.Run())

Ek: Geçici erişim kimlik bilgilerini kullanarak bir TunnelClient örneğini başlatmaya ilişkin örnek kod

import (
    otscommon "github.com/aliyun/aliyun-tablestore-go-sdk/common"
    "github.com/aliyun/aliyun-tablestore-go-sdk/tunnel"
    "sync"
    "time"
)

type RefreshClient struct {
    lastRefresh          time.Time
    refreshIntervalInMin int
}

func NewRefreshClient(intervalInMin int) *RefreshClient {
    return &RefreshClient{
        refreshIntervalInMin: intervalInMin,
    }
}

func (c *RefreshClient) IsExpired() bool {
    now := time.Now()
    if c.lastRefresh.IsZero() || now.Sub(c.lastRefresh) > time.Duration(c.refreshIntervalInMin)*time.Minute {
        return true
    }

    return false
}

func (c *RefreshClient) Update() {
    c.lastRefresh = time.Now()
}

type clientCredentials struct {
    accessKeyID     string
    accessKeySecret string
    securityToken   string
}

func newClientCredentials(accessKeyID string, accessKeySecret string, securityToken string) *clientCredentials {
    return &clientCredentials{accessKeyID: accessKeyID, accessKeySecret: accessKeySecret, securityToken: securityToken}
}

func (c *clientCredentials) GetAccessKeyID() string {
    return c.accessKeyID
}

func (c *clientCredentials) GetAccessKeySecret() string {
    return c.accessKeySecret
}

func (c *clientCredentials) GetSecurityToken() string {
    return c.securityToken
}

type OTSCredentialsProvider struct {
    refresh *RefreshClient
    cred    *clientCredentials
    lock    sync.Mutex
}

func NewOTSCredentialsProvider() *OTSCredentialsProvider {
    return &OTSCredentialsProvider{
        // Modify the refresh cycle for temporary access credentials based on your business requirements. The refresh cycle must be shorter than the validity period of the temporary access credentials. 
        refresh: NewRefreshClient(30),
    }
}

func (p *OTSCredentialsProvider) renewCredentials() error {
    if p.cred == nil || p.refresh.IsExpired() {
        // Obtain temporary access credentials. You can call the AssumeRole operation of RAM to obtain the AccessKey ID, AccessKey secret, security token, and validity period of the temporary access credentials. 
        // Configure the following parameters. For information about RAM SDKs, see the documentation of RAM.  
        // resp, err := GetUserOtsStsToken()
        accessKeyId := ""
        accessKeySecret := ""
        stsToken := ""
        p.cred = newClientCredentials(accessKeyId, accessKeySecret, stsToken)
        p.refresh.Update()
    }

    return nil
}

func (p *OTSCredentialsProvider) GetCredentials() otscommon.Credentials {
    p.lock.Lock()
    defer p.lock.Unlock()

    if err := p.renewCredentials(); err != nil {
        // log error
        if p.cred == nil {
            return newClientCredentials("", "", "")
        }
    }

    return p.cred
}

// NewTunnelClientWithToken is used to initialize a TunnelClient instance with the feature of refreshing temporary access credentials. 
func NewTunnelClientWithToken(endpoint, instanceName, accessId, accessKey, token string) tunnel.TunnelClient {
    return tunnel.NewTunnelClientWithToken(
        endpoint,
        instanceName,
        "",
        "",
        "",
        nil,
        tunnel.SetCredentialsProvider(NewOTSCredentialsProvider()),
    )
}