如何使用 Go 跟踪分段上传到 s3 的进度？

Question

我正在尝试使用 Mitchell Hashimoto 的 goamz fork 提供的 PutPart 方法。可悲的是，每次我拿回一个零件并检查它的大小时，它似乎认为它是整个文件的大小，而不仅仅是一个块。

例如

上传我希望看到的 15m 文件时

Uploading...
Processing 1 part of 3 and uploaded 5242880.0 bytes.
 Processing 2 part of 3 and uploaded 5242880.0 bytes.
 Processing 3 part of 3 and uploaded 5242880.0 bytes.

相反，我看到了：

Uploading...
Processing 1 part of 3 and uploaded 15728640 bytes.
 Processing 2 part of 3 and uploaded 15728640 bytes.
 Processing 3 part of 3 and uploaded 15728640 bytes.

这是由于 file.Read(partBuffer) 的问题吗？任何帮助将非常感激。

我在 mac 上使用 go 1.5.1。

package main

import (
    "bufio"
    "fmt"
    "math"
    "net/http"
    "os"

    "github.com/mitchellh/goamz/aws"
    "github.com/mitchellh/goamz/s3"
)

func check(err error) {
    if err != nil {
        panic(err)
    }
}

func main() {
    fmt.Println("Test")

    auth, err := aws.GetAuth("XXXXX", "XXXXXXXXXX")
    check(err)

    client := s3.New(auth, aws.USWest2)

    b := s3.Bucket{
        S3:   client,
        Name: "some-bucket",
    }

    fileToBeUploaded := "testfile"
    file, err := os.Open(fileToBeUploaded)
    check(err)
    defer file.Close()

    fileInfo, _ := file.Stat()
    fileSize := fileInfo.Size()
    bytes := make([]byte, fileSize)

    // read into buffer
    buffer := bufio.NewReader(file)
    _, err = buffer.Read(bytes)
    check(err)
    filetype := http.DetectContentType(bytes)

    // set up for multipart upload
    multi, err := b.InitMulti("/"+fileToBeUploaded, filetype, s3.ACL("bucket-owner-read"))
    check(err)

    const fileChunk = 5242880 // 5MB
    totalPartsNum := uint64(math.Ceil(float64(fileSize) / float64(fileChunk)))
    parts := []s3.Part{}

    fmt.Println("Uploading...")
    for i := uint64(1); i < totalPartsNum; i++ {

        partSize := int(math.Min(fileChunk, float64(fileSize-int64(i*fileChunk))))
        partBuffer := make([]byte, partSize)

        _, err := file.Read(partBuffer)
        check(err)

        part, err := multi.PutPart(int(i), file) // write to S3 bucket part by part
        check(err)

        fmt.Printf("Processing %d part of %d and uploaded %d bytes.\n ", int(i), int(totalPartsNum), int(part.Size))
        parts = append(parts, part)
    }

    err = multi.Complete(parts)
    check(err)

    fmt.Println("\n\nPutPart upload completed")

}

Answer 1

这里的问题可能是由于没有完全读取文件引起的。 Read 可能有点微妙：

Read 最多将 len(p) 个字节读入 p。它返回读取的字节数 (0

所以您可能应该使用ioReadFull 或（更好）io.CopyN。

也就是说，我认为您应该尝试切换到官方 AWS Go 软件包。他们有一个方便的Uploader 会为您处理所有这些：

package main

import (
    "log"
    "os"

    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/s3/s3manager"
)

func main() {
    bucketName := "test-bucket"
    keyName := "test-key"
    file, err := os.Open("example")
    if err != nil {
        log.Fatalln(err)
    }
    defer file.Close()

    sess := session.New()
    uploader := s3manager.NewUploader(sess)

    // Perform an upload.
    result, err := uploader.Upload(&s3manager.UploadInput{
        Bucket: &bucketName,
        Key:    &keyName,
        Body:   file,
    })
    if err != nil {
        log.Fatalln(err)
    }
    log.Println(result)
}

您可以在 godoc.org 上找到更多文档。

Answer 2

您读入partBuffer 的数据根本没有被使用。您将file 传递给multi.PutPart，它会读取file 的全部内容，并在必要时将其返回到开头并清除您所做的所有工作。 p>

对代码的最小更改是将bytes.NewReader(partBuffer) 传递给PutPart，而不是file。 bytes.Reader 实现了PutPart 需要的io.ReadSeeker 接口，并且还将报告其大小为partBuffer 的大小。

另一种方法是使用 io.SectionReader 类型 — 无需自己将数据读入缓冲区，您只需基于 file 创建一系列 SectionReaders，并使用您想要并传递的大小和偏移量它们进入PutPart，它们会将读取传递给底层文件读取器。这应该同样有效，并大大减少了您必须编写（和错误检查）的代码。它还避免了在 RAM 中不必要地缓冲整个数据块。

Answer 3

当您将文件部分传递给 multi.PutPart 方法 (n, strings.NewReader ("")) 时，您的代码将不得不更改一些点这要正常工作，下面的代码就可以工作了。

记住 PutPart 发送分段上传的一部分，从 r 读取所有内容，每个部分，除了最后一个，必须大小至少为 5MB。 它在 goamz 文档中有所描述。

我已更改为正常工作的点：

在这里，我正在使用文件的所有字节创建我们的 headerPart

HeaderPart: = strings.NewReader（字符串（字节））

这里 io.ReadFull (HeaderPart, partBuffer) 我正在读取 make ([] byte, partSize) 命令的整个缓冲区部分，每次都是定位在文件的某个部分。

当我们运行 multi.PutPart (int (i) +1, strings.NewReader (string (partBuffer))) 时，我们必须 +1，因为它不计算部分 0，而是在传递目标文件时，我们将为此使用 strings.NewReader 函数传递部分内容。

查看下面的代码，它现在可以正常工作了。

package main

import(
"bufio"
"fmt"
"math"
"net/http"
"os"
"launchpad.net/goamz/aws"
"launchpad.net/goamz/s3"
)

func check(err error) {
    if err != nil {
     panic(err)
    } 
}

func main() {

fmt.Println("Test")

auth := aws.Auth{
    AccessKey: "xxxxxxxxxxx", // change this to yours
    SecretKey: "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
}

client := s3.New(auth, aws.USWest2)

b := s3.Bucket{
    S3:   client,
    Name: "some-bucket",
}

fileToBeUploaded := "testfile"
file, err := os.Open(fileToBeUploaded)
check(err)
defer file.Close()

fileInfo, _ := file.Stat()
fileSize := fileInfo.Size()
bytes := make([]byte, fileSize)

// read into buffer
buffer := bufio.NewReader(file)
_, err = buffer.Read(bytes)
check(err)
filetype := http.DetectContentType(bytes)

// set up for multipart upload
multi, err := b.InitMulti("/"+fileToBeUploaded, filetype, s3.ACL("bucket-owner-read"))
check(err)

const fileChunk = 5242880 // 5MB
totalPartsNum := uint64(math.Ceil(float64(fileSize) / float64(fileChunk)))
parts := []s3.Part{}

fmt.Println("Uploading...")

HeaderPart := strings.NewReader(string(bytes))

for i := uint64(0); i < totalPartsNum; i++ {

    partSize := int(math.Min(fileChunk, float64(fileSize-int64(i*fileChunk))))

    partBuffer := make([]byte, partSize)

    n , errx := io.ReadFull(HeaderPart, partBuffer)

    check(errx)

    part, err := multi.PutPart(int(i)+1, strings.NewReader(string(partBuffer))) // write to S3 bucket part by part

    check(err)

    fmt.Printf("Processing %d part of %d and uploaded %d bytes.\n ", int(i), int(totalPartsNum), int(n))
    parts = append(parts, part)
}

err = multi.Complete(parts)
check(err)

fmt.Println("\n\nPutPart upload completed")
}