下载大文件时,可以使用 DownLoadFile 断点续传下载接口。断点续传下载使用了分片下载将内容写入到临时文件中,如果出现部分分片成功部分失败,可以记录下载成功和失败分片的信息,保存到 Checkpoint 文件中。再次下载相同对象时,只下载失败的分片。当分片全部成功时将临时文件重命名为正式文件,并删除 Checkpoint 文件。
tos:GetObject
权限,具体操作,请参见权限配置指南。tos:GetObjectVersion
权限,具体操作,请参见权限配置指南。Checkpoint
文件中,所以程序需要对 Checkpoint
文件有写入权限。Checkpoint
文件中,如果下载过程中某一分片下载失败,再次下载时会从 Checkpoint
文件中记录的点继续下载,从而达到断点续传的效果。下载完成后, Checkpoint
文件会被删除。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) }