You need to enable JavaScript to run this app.
导航
对象存储
  • 文档首页
    • /
  • 对象存储

断点续传下载（Go SDK）

最近更新时间：2024.04.22 16:09:55

首次发布时间：2022.09.27 11:45:29

我的收藏

下载大文件时，可以使用 DownLoadFile 断点续传下载接口。断点续传下载使用了分片下载将内容写入到临时文件中，如果出现部分分片成功部分失败，可以记录下载成功和失败分片的信息，保存到 Checkpoint 文件中。再次下载相同对象时，只下载失败的分片。当分片全部成功时将临时文件重命名为正式文件，并删除 Checkpoint 文件。

注意事项

  • 下载对象前，您必须具有 tos:GetObject 权限，具体操作，请参见权限配置指南。
  • 对于开启多版本的桶，下载指定版本对象时，您必须具有 tos:GetObjectVersion 权限，具体操作，请参见权限配置指南。
  • 如果应用程序会在同一时刻大量下载同一个对象，您的访问速度会受到 TOS 带宽及地域的限制。建议您使用 CDN 产品，提升性能的同时也能降低您的成本。通过 CDN 访问 TOS 的详细信息，请参见使用 CDN 加速访问 TOS 资源。
  • SDK 会将下载的状态信息记录在 Checkpoint 文件中，所以程序需要对 Checkpoint 文件有写入权限。
  • 使用断点续传下载时，文件下载的进度信息会记录在 Checkpoint 文件中，如果下载过程中某一分片下载失败，再次下载时会从 Checkpoint 文件中记录的点继续下载，从而达到断点续传的效果。下载完成后， Checkpoint 文件会被删除。
  • 如果下载过程中文件的 ETag 发生变化、Part 丢失或被修改，则重新下载文件。

示例代码

断点续传下载

package main

import (
   "context"
   "fmt"
   "sync"
   "time"

   "github.com/volcengine/ve-tos-golang-sdk/v2/tos"
   "github.com/volcengine/ve-tos-golang-sdk/v2/tos/enum"
)

// 自定义事件监听，需要实现 tos.DownloadEventListener 接口
type eventChange struct {
}

func (e eventChange) EventChange(event *tos.DownloadEvent) {
   switch event.Type {
   case enum.DownloadEventCreateTempFileSucceed:
      fmt.Println("Download File Create Temp File Success.")
   case enum.DownloadEventCreateTempFileFailed:
      fmt.Println("Download File Create Temp File Fail.")
   case enum.DownloadEventDownloadPartSucceed:
      fmt.Printf("Download %s %s part success, DownloadPartInfo:%v\n", event.Bucket, event.Key, event.DowloadPartInfo)
   case enum.DownloadEventDownloadPartAborted:
      fmt.Printf("Download %s %s part aborted.", event.Bucket, event.Key)
   case enum.DownloadEventDownloadPartFailed:
      fmt.Printf("Download %s %s part fail, err:%v\n", event.Bucket, event.Key, event.Err)
   case enum.DownloadEventRenameTempFileSucceed:
      fmt.Printf("Download %s %s success.\n", event.Bucket, event.Key)
   case enum.DownloadEventRenameTempFileFailed:
      fmt.Printf("Download %s %s fail,err:%v\n", event.Bucket, event.Key, event.Err)

   }
}

// 自定义进度回调，需要实现 tos.DataTransferStatusChange 接口
type listener struct {
}

func (l *listener) DataTransferStatusChange(event *tos.DataTransferStatus) {
   switch event.Type {
   case enum.DataTransferStarted:
      fmt.Println("Data transfer started")
   case enum.DataTransferRW:
      if event.TotalBytes != -1 {
         fmt.Printf("Once Read:%d,ConsumerBytes/TotalBytes: %d/%d,%d%%\n", event.RWOnceBytes, event.ConsumedBytes, event.TotalBytes, event.ConsumedBytes*100/event.TotalBytes)
      } else {
         fmt.Printf("Once Read:%d,ConsumerBytes:%d\n", event.RWOnceBytes, event.ConsumedBytes)
      }
   case enum.DataTransferSucceed:
      fmt.Printf("Data Transfer Succeed, ConsumerBytes/TotalBytes: %d/%d,%d%%\n", event.ConsumedBytes, event.TotalBytes, event.ConsumedBytes*100/event.TotalBytes)
   case enum.DataTransferFailed:
      fmt.Printf("Data Transfer Failed\n")
   }
}

func checkErr(err error) {
   if err != nil {
      if serverErr, ok := err.(*tos.TosServerError); ok {
         fmt.Println("Error:", serverErr.Error())
         fmt.Println("Request ID:", serverErr.RequestID)
         fmt.Println("Response Status Code:", serverErr.StatusCode)
         fmt.Println("Response Header:", serverErr.Header)
         fmt.Println("Response Err Code:", serverErr.Code)
         fmt.Println("Response Err Msg:", serverErr.Message)
      } else if clientErr, ok := err.(*tos.TosClientError); ok {
         fmt.Println("Error:", clientErr.Error())
         fmt.Println("Client Cause Err:", clientErr.Cause.Error())
      } else {
         fmt.Println("Error:", err)
      }
      panic(err)
   }
}

type rateLimit struct {
   rate     int64
   capacity int64

   currentAmount int64
   sync.Mutex
   lastConsumeTime time.Time
}

func NewDefaultRateLimit(rate int64, capacity int64) tos.RateLimiter {
   return &rateLimit{
      rate:            rate,
      capacity:        capacity,
      lastConsumeTime: time.Now(),
      currentAmount:   capacity,
      Mutex:           sync.Mutex{},
   }
}

func (d *rateLimit) Acquire(want int64) (ok bool, timeToWait time.Duration) {
   d.Lock()
   defer d.Unlock()
   if want > d.capacity {
      want = d.capacity
   }
   increment := int64(time.Now().Sub(d.lastConsumeTime).Seconds() * float64(d.rate))
   if increment+d.currentAmount > d.capacity {
      d.currentAmount = d.capacity
   } else {
      d.currentAmount += increment
   }
   if want > d.currentAmount {
      timeToWaitSec := float64(want-d.currentAmount) / float64(d.rate)
      return false, time.Duration(timeToWaitSec * float64(time.Second))
   }
   d.lastConsumeTime = time.Now()
   d.currentAmount -= want
   return true, 0
}

func main() {
   var (
      accessKey = os.Getenv("TOS_ACCESS_KEY")
      secretKey = os.Getenv("TOS_SECRET_KEY")
      // Bucket 对应的 Endpoint，以华北2（北京）为例：https://tos-cn-beijing.volces.com
      endpoint = "https://tos-cn-beijing.volces.com"
      region   = "cn-beijing"
      // 填写 BucketName
      bucketName = "*** Provide your bucket name ***"

      // 下载指定的对象
      objectKey = "example_dir/example.txt"
      ctx       = context.Background()
   )
   // 初始化客户端
   client, err := tos.NewClientV2(endpoint, tos.WithRegion(region), tos.WithCredentials(tos.NewStaticCredentials(accessKey, secretKey)))
   checkErr(err)
   // 下载文件到指定的路径
   cancel := tos.NewCancelHook()
   rateLimit1m := int64(1024 * 1024)
   output, err := client.DownloadFile(ctx, &tos.DownloadFileInput{
      HeadObjectV2Input: tos.HeadObjectV2Input{
         Bucket: bucketName,
         Key:    objectKey,
      },
      // 下载的文件路径
      FilePath: "./example.txt",
      // 下载时指定分片大小
      PartSize: tos.DefaultPartSize,
      // 分片下载任务并发数量
      TaskNum: 5,
      // 开启断点续传
      EnableCheckpoint: true,
      // 指定断点续传临时文件路径
      CheckpointFile: "./example.checkpoint",
      // 数据传输回调
      DataTransferListener: &listener{},
      // 事件监听回调
      DownloadEventListener: eventChange{},
      // 下载客户端限速
      RateLimiter: NewDefaultRateLimit(rateLimit1m, rateLimit1m),
      // 取消下载
      CancelHook: cancel,
   })
   checkErr(err)
   fmt.Println("DownloadFile Request ID:", output.RequestID)
}

配置进度条

断点续传下载时可通过实现 tos.DataTransferStatusChange 接口接收下载进度，代码示例如下。

package main

import (
   "context"
   "fmt"

   "github.com/volcengine/ve-tos-golang-sdk/v2/tos"
   "github.com/volcengine/ve-tos-golang-sdk/v2/tos/enum"
)

// 自定义进度回调，需要实现 tos.DataTransferStatusChange 接口
type listener struct {
}

func (l *listener) DataTransferStatusChange(event *tos.DataTransferStatus) {
   switch event.Type {
   case enum.DataTransferStarted:
      fmt.Println("Data transfer started")
   case enum.DataTransferRW:
      if event.TotalBytes != -1 {
         fmt.Printf("Once Read:%d,ConsumerBytes/TotalBytes: %d/%d,%d%%\n", event.RWOnceBytes, event.ConsumedBytes, event.TotalBytes, event.ConsumedBytes*100/event.TotalBytes)
      } else {
         fmt.Printf("Once Read:%d,ConsumerBytes:%d\n", event.RWOnceBytes, event.ConsumedBytes)
      }
   case enum.DataTransferSucceed:
      fmt.Printf("Data Transfer Succeed, ConsumerBytes/TotalBytes: %d/%d,%d%%\n", event.ConsumedBytes, event.TotalBytes, event.ConsumedBytes*100/event.TotalBytes)
   case enum.DataTransferFailed:
      fmt.Printf("Data Transfer Failed\n")
   }
}

func checkErr(err error) {
   if err != nil {
      if serverErr, ok := err.(*tos.TosServerError); ok {
         fmt.Println("Error:", serverErr.Error())
         fmt.Println("Request ID:", serverErr.RequestID)
         fmt.Println("Response Status Code:", serverErr.StatusCode)
         fmt.Println("Response Header:", serverErr.Header)
         fmt.Println("Response Err Code:", serverErr.Code)
         fmt.Println("Response Err Msg:", serverErr.Message)
      } else if clientErr, ok := err.(*tos.TosClientError); ok {
         fmt.Println("Error:", clientErr.Error())
         fmt.Println("Client Cause Err:", clientErr.Cause.Error())
      } else {
         fmt.Println("Error:", err)
      }
      panic(err)
   }
}

func main() {
   var (
      accessKey = os.Getenv("TOS_ACCESS_KEY")
      secretKey = os.Getenv("TOS_SECRET_KEY")
      // Bucket 对应的 Endpoint，以华北2（北京）为例：https://tos-cn-beijing.volces.com
      endpoint = "https://tos-cn-beijing.volces.com"
      region   = "cn-beijing"
      // 填写 BucketName
      bucketName = "*** Provide your bucket name ***"

      // 下载指定的对象
      objectKey = "example_dir/example.txt"
      ctx       = context.Background()
   )
   // 初始化客户端
   client, err := tos.NewClientV2(endpoint, tos.WithRegion(region), tos.WithCredentials(tos.NewStaticCredentials(accessKey, secretKey)))
   checkErr(err)
   // 下载文件到指定的路径
   output, err := client.DownloadFile(ctx, &tos.DownloadFileInput{
      HeadObjectV2Input: tos.HeadObjectV2Input{
         Bucket: bucketName,
         Key:    objectKey,
      },
      // 下载的文件路径
      FilePath: "./example.txt",
      // 下载时指定分片大小
      PartSize: tos.DefaultPartSize,
      // 分片下载任务并发数量
      TaskNum: 5,
      // 开启断点续传
      EnableCheckpoint: true,
      // 指定断点续传临时文件路径
      CheckpointFile: "./example.checkpoint",
      // 数据传输回调
      DataTransferListener: &listener{},
   })
   checkErr(err)
   fmt.Println("DownloadFile Request ID:", output.RequestID)
}

配置事件回调

以下代码用于自定义断点续传下载回调函数。

package main

import (
   "context"
   "fmt"

   "github.com/volcengine/ve-tos-golang-sdk/v2/tos"
   "github.com/volcengine/ve-tos-golang-sdk/v2/tos/enum"
)

// 自定义事件监听，需要实现 tos.UploadEventListener 接口
type eventChange struct {
}

func (e eventChange) EventChange(event *tos.DownloadEvent) {
   switch event.Type {
   case enum.DownloadEventCreateTempFileSucceed:
      fmt.Println("Download File Create Temp File Success.")
   case enum.DownloadEventCreateTempFileFailed:
      fmt.Println("Download File Create Temp File Fail.")
   case enum.DownloadEventDownloadPartSucceed:
      fmt.Printf("Download %s %s part success, DownloadPartInfo:%v\n", event.Bucket, event.Key, event.DowloadPartInfo)
   case enum.DownloadEventDownloadPartAborted:
      fmt.Printf("Download %s %s part aborted.", event.Bucket, event.Key)
   case enum.DownloadEventDownloadPartFailed:
      fmt.Printf("Download %s %s part fail, err:%v\n", event.Bucket, event.Key, event.Err)
   case enum.DownloadEventRenameTempFileSucceed:
      fmt.Printf("Download %s %s success.\n", event.Bucket, event.Key)
   case enum.DownloadEventRenameTempFileFailed:
      fmt.Printf("Download %s %s fail,err:%v\n", event.Bucket, event.Key, event.Err)

   }
}

func checkErr(err error) {
   if err != nil {
      if serverErr, ok := err.(*tos.TosServerError); ok {
         fmt.Println("Error:", serverErr.Error())
         fmt.Println("Request ID:", serverErr.RequestID)
         fmt.Println("Response Status Code:", serverErr.StatusCode)
         fmt.Println("Response Header:", serverErr.Header)
         fmt.Println("Response Err Code:", serverErr.Code)
         fmt.Println("Response Err Msg:", serverErr.Message)
      } else if clientErr, ok := err.(*tos.TosClientError); ok {
         fmt.Println("Error:", clientErr.Error())
         fmt.Println("Client Cause Err:", clientErr.Cause.Error())
      } else {
         fmt.Println("Error:", err)
      }
      panic(err)
   }
}

func main() {
   var (
      accessKey = os.Getenv("TOS_ACCESS_KEY")
      secretKey = os.Getenv("TOS_SECRET_KEY")
      // Bucket 对应的 Endpoint，以华北2（北京）为例：https://tos-cn-beijing.volces.com
      endpoint = "https://tos-cn-beijing.volces.com"
      region   = "cn-beijing"
      // 填写 BucketName
      bucketName = "*** Provide your bucket name ***"

      // 下载指定的对象
      objectKey = "example_dir/example.txt"
      ctx       = context.Background()
   )
   // 初始化客户端
   client, err := tos.NewClientV2(endpoint, tos.WithRegion(region), tos.WithCredentials(tos.NewStaticCredentials(accessKey, secretKey)))
   checkErr(err)
   // 下载文件到指定的路径
   cancel := tos.NewCancelHook()
   output, err := client.DownloadFile(ctx, &tos.DownloadFileInput{
      HeadObjectV2Input: tos.HeadObjectV2Input{
         Bucket: bucketName,
         Key:    objectKey,
      },
      // 下载的文件路径
      FilePath: "./example.txt",
      // 下载时指定分片大小
      PartSize: tos.DefaultPartSize,
      // 分片下载任务并发数量
      TaskNum: 5,
      // 开启断点续传
      EnableCheckpoint: true,
      // 事件监听回调
      DownloadEventListener: eventChange{},
      // 取消下载
      CancelHook: cancel,
   })
   checkErr(err)
   fmt.Println("DownloadFile Request ID:", output.RequestID)
}

配置客户端限速

断点续传下载时可以通过客户端使用 tos.RateLimiter 接口对下载数据所占用的带宽进行限制，代码如下所示。

package main

import (
   "context"
   "fmt"
   "sync"
   "time"

   "github.com/volcengine/ve-tos-golang-sdk/v2/tos"
)

func checkErr(err error) {
   if err != nil {
      if serverErr, ok := err.(*tos.TosServerError); ok {
         fmt.Println("Error:", serverErr.Error())
         fmt.Println("Request ID:", serverErr.RequestID)
         fmt.Println("Response Status Code:", serverErr.StatusCode)
         fmt.Println("Response Header:", serverErr.Header)
         fmt.Println("Response Err Code:", serverErr.Code)
         fmt.Println("Response Err Msg:", serverErr.Message)
      } else if clientErr, ok := err.(*tos.TosClientError); ok {
         fmt.Println("Error:", clientErr.Error())
         fmt.Println("Client Cause Err:", clientErr.Cause.Error())
      } else {
         fmt.Println("Error:", err)
      }
      panic(err)
   }
}

type rateLimit struct {
   rate     int64
   capacity int64

   currentAmount int64
   sync.Mutex
   lastConsumeTime time.Time
}

func NewDefaultRateLimit(rate int64, capacity int64) tos.RateLimiter {
   return &rateLimit{
      rate:            rate,
      capacity:        capacity,
      lastConsumeTime: time.Now(),
      currentAmount:   capacity,
      Mutex:           sync.Mutex{},
   }
}

func (d *rateLimit) Acquire(want int64) (ok bool, timeToWait time.Duration) {
   d.Lock()
   defer d.Unlock()
   if want > d.capacity {
      want = d.capacity
   }
   increment := int64(time.Now().Sub(d.lastConsumeTime).Seconds() * float64(d.rate))
   if increment+d.currentAmount > d.capacity {
      d.currentAmount = d.capacity
   } else {
      d.currentAmount += increment
   }
   if want > d.currentAmount {
      timeToWaitSec := float64(want-d.currentAmount) / float64(d.rate)
      return false, time.Duration(timeToWaitSec * float64(time.Second))
   }
   d.lastConsumeTime = time.Now()
   d.currentAmount -= want
   return true, 0
}

func main() {
   var (
      accessKey = os.Getenv("TOS_ACCESS_KEY")
      secretKey = os.Getenv("TOS_SECRET_KEY")
      // Bucket 对应的 Endpoint，以华北2（北京）为例：https://tos-cn-beijing.volces.com
      endpoint = "https://tos-cn-beijing.volces.com"
      region   = "cn-beijing"
      // 填写 BucketName
      bucketName = "*** Provide your bucket name ***"

      // 下载指定的对象
      objectKey = "example_dir/example.txt"
      ctx       = context.Background()
   )
   // 初始化客户端
   client, err := tos.NewClientV2(endpoint, tos.WithRegion(region), tos.WithCredentials(tos.NewStaticCredentials(accessKey, secretKey)))
   checkErr(err)
   // 下载文件到指定的路径
   rateLimit1m := int64(1024 * 1024)
   output, err := client.DownloadFile(ctx, &tos.DownloadFileInput{
      HeadObjectV2Input: tos.HeadObjectV2Input{
         Bucket: bucketName,
         Key:    objectKey,
      },
      // 下载的文件路径
      FilePath: "./example.txt",
      // 下载时指定分片大小
      PartSize: tos.DefaultPartSize,
      // 分片下载任务并发数量
      TaskNum: 5,
      // 开启断点续传
      EnableCheckpoint: true,
      // 指定断点续传临时文件路径
      CheckpointFile: "./example.checkpoint",
      // 下载客户端限速
      RateLimiter: NewDefaultRateLimit(rateLimit1m, rateLimit1m),
   })
   checkErr(err)
   fmt.Println("DownloadFile Request ID:", output.RequestID)
}

取消机制

以下代码用于在运行时取消正在执行的断点续传下载任务。

package main

import (
   "context"
   "fmt"

   "github.com/volcengine/ve-tos-golang-sdk/v2/tos"
)

func checkErr(err error) {
   if err != nil {
      if serverErr, ok := err.(*tos.TosServerError); ok {
         fmt.Println("Error:", serverErr.Error())
         fmt.Println("Request ID:", serverErr.RequestID)
         fmt.Println("Response Status Code:", serverErr.StatusCode)
         fmt.Println("Response Header:", serverErr.Header)
         fmt.Println("Response Err Code:", serverErr.Code)
         fmt.Println("Response Err Msg:", serverErr.Message)
      } else if clientErr, ok := err.(*tos.TosClientError); ok {
         fmt.Println("Error:", clientErr.Error())
         fmt.Println("Client Cause Err:", clientErr.Cause.Error())
      } else {
         fmt.Println("Error:", err)
      }
      panic(err)
   }
}

func main() {
   var (
      accessKey = os.Getenv("TOS_ACCESS_KEY")
      secretKey = os.Getenv("TOS_SECRET_KEY")
      // Bucket 对应的 Endpoint，以华北2（北京）为例：https://tos-cn-beijing.volces.com
      endpoint = "https://tos-cn-beijing.volces.com"
      region   = "cn-beijing"
      // 填写 BucketName
      bucketName = "*** Provide your bucket name ***"

      // 下载指定的对象
      objectKey = "example_dir/example.txt"
      ctx       = context.Background()
   )
   // 初始化客户端
   client, err := tos.NewClientV2(endpoint, tos.WithRegion(region), tos.WithCredentials(tos.NewStaticCredentials(accessKey, secretKey)))
   checkErr(err)
   // 下载文件到指定的路径
   cancel := tos.NewCancelHook()
   output, err := client.DownloadFile(ctx, &tos.DownloadFileInput{
      HeadObjectV2Input: tos.HeadObjectV2Input{
         Bucket: bucketName,
         Key:    objectKey,
      },
      // 下载的文件路径
      FilePath: "./example.txt",
      // 下载时指定分片大小
      PartSize: tos.DefaultPartSize,
      // 分片下载任务并发数量
      TaskNum: 5,
      // 开启断点续传
      EnableCheckpoint: true,
      // 指定断点续传临时文件路径
      CheckpointFile: "./example.checkpoint",
      // 取消下载
      CancelHook: cancel,
   })
   checkErr(err)
   fmt.Println("DownloadFile Request ID:", output.RequestID)
}