如何使用Golang在中断后恢复GCS大文件上传？

问题背景 #

使用Go语言的cloud.google.com/go/storage客户端上传GB级大文件到Google Cloud Storage时，小文件上传正常，但大文件上传过程中笔记本休眠唤醒后会触发连接重置错误：

read tcp x.x.x.x:ppp->X.X.X.X:PPP: read: connection reset by peer

需要实现断点续传，避免从头重试上传。

实现步骤 #

GCS官方支持可恢复上传，Go客户端通过ResumableWriter实现该功能，核心是保存并复用恢复令牌（Resume Token），以下是具体实现代码：

1. 核心上传逻辑 #

package main

import (
	"io"
	"log"
	"os"

	"cloud.google.com/go/storage"
	"golang.org/x/net/context"
)

func uploadToGCS(ctx context.Context, bucket, fileName string, r io.Reader) error {
	client, err := storage.NewClient(ctx)
	if err != nil {
		return err
	}
	defer client.Close()

	obj := client.Bucket(bucket).Object(fileName)
	var w *storage.Writer

	// 尝试加载之前保存的恢复令牌
	resumeToken, err := loadResumeToken(fileName)
	if err != nil {
		return err
	}

	if resumeToken == "" {
		// 无令牌，创建新的可恢复上传Writer
		w = obj.NewResumableWriter(ctx)
	} else {
		// 使用令牌恢复上传
		w = obj.NewResumableWriterFromToken(ctx, resumeToken)
	}

	// 设置文件属性（和原代码保持一致）
	w.ContentType = ""
	w.ACL = []storage.ACLRule{{Entity: storage.AllUsers, Role: storage.RoleReader}}

	// 定位到已上传的位置，跳过已传输的字节
	if w.ResumeToken != "" {
		uploadedBytes, err := w.Seek(0, io.SeekEnd)
		if err != nil {
			return err
		}

		// 如果读取器支持Seek（比如本地文件），直接定位
		if seeker, ok := r.(io.Seeker); ok {
			if _, err := seeker.Seek(uploadedBytes, io.SeekStart); err != nil {
				return err
			}
		} else {
			// 不支持Seek则丢弃已上传的字节
			if _, err := io.CopyN(io.Discard, r, uploadedBytes); err != nil {
				return err
			}
		}
	}

	// 上传过程中出错时保存令牌
	defer func() {
		if err != nil && w.ResumeToken != "" {
			if saveErr := saveResumeToken(fileName, w.ResumeToken); saveErr != nil {
				log.Printf("警告：保存恢复令牌失败: %v", saveErr)
			}
		}
	}()

	// 执行上传
	if _, err := io.Copy(w, r); err != nil {
		return err
	}

	// 上传成功，删除本地保存的令牌
	if err := deleteResumeToken(fileName); err != nil {
		log.Printf("警告：删除恢复令牌失败: %v", err)
	}

	// 完成Writer关闭
	return w.Close()
}

2. 恢复令牌的持久化 #

需要实现令牌的本地保存、读取和删除逻辑，这里以本地文件为例：

// saveResumeToken 将恢复令牌保存到本地文件
func saveResumeToken(fileName, token string) error {
	tokenPath := fileName + ".resume"
	return os.WriteFile(tokenPath, []byte(token), 0644)
}

// loadResumeToken 从本地文件加载恢复令牌
func loadResumeToken(fileName string) (string, error) {
	tokenPath := fileName + ".resume"
	data, err := os.ReadFile(tokenPath)
	if err != nil {
		if os.IsNotExist(err) {
			return "", nil // 文件不存在返回空令牌，不报错
		}
		return "", err
	}
	return string(data), nil
}

// deleteResumeToken 删除本地保存的恢复令牌
func deleteResumeToken(fileName string) error {
	tokenPath := fileName + ".resume"
	return os.Remove(tokenPath)
}

3. 关键注意事项 #

• 读取器兼容性：如果上传的数据源不是本地文件（不支持io.Seeker），需要通过io.CopyN跳过已上传的字节，确保从断点开始传输。
• 令牌更新：上传过程中令牌可能会变化，出错时务必保存当前最新的令牌。
• 清理令牌：上传完成后必须删除本地令牌，避免下次上传错误复用旧令牌。
• 错误处理：令牌的保存/删除操作建议使用日志警告，不要中断主流程。

内容的提问来源于stack exchange，提问作者Ryan