能否在 golang 中实现数据流式传输？

积累知识，胜过积蓄金银！毕竟在Golang开发的过程中，会遇到各种各样的问题，往往都是一些细节知识点还没有掌握好而导致的，因此基础知识点的积累是很重要的。下面本文《能否在 golang 中实现数据流式传输？》，就带大家讲解一下知识点，若是你对本文感兴趣，或者是想搞懂其中某个知识点，就请你继续往下看吧~

我想要处理一些内容不适合我的工作人员内存的文件。到目前为止我找到的解决方案涉及将处理结果保存到 /tmp 目录，然后将其上传到 s3。

import (
    "bufio"
    "bytes"
    "context"
    "fmt"
    "log"
    "os"
    "runtime"
    "strings"
    "sync"

    "github.com/aws/aws-sdk-go-v2/service/s3"
    "github.com/korovkin/limiter"
    "github.com/xitongsys/parquet-go/parquet"
    "github.com/xitongsys/parquet-go/writer"
)

func DownloadWarc(
    ctx context.Context,
    s3Client *s3.Client,
    warcs []*types.Warc,
    path string,
) error {
    key := fmt.Sprintf("parsed_warc/%s.parquet", path)

    filename := fmt.Sprintf("/tmp/%s", path)
    file, err := os.Create(filename)
    if err != nil {
        return fmt.Errorf("error creating file: %s", err)
    }
    defer file.Close()

    bytesWriter := bufio.NewWriter(file)
    pw, err := writer.NewParquetWriterFromWriter(bytesWriter, new(Page), 4)
    if err != nil {
        return fmt.Errorf("Can't create parquet writer: %s", err)
    }

    pw.RowGroupSize = 128 * 1024 * 1024 //128M
    pw.CompressionType = parquet.CompressionCodec_SNAPPY

    mutex := sync.Mutex{}
    numWorkers := runtime.NumCPU() * 2
    fmt.Printf("Using %d workers\n", numWorkers)
    limit := limiter.NewConcurrencyLimiter(numWorkers)

    for i, warc := range warcs {
        limit.Execute(func() {
            log.Printf("%d: %+v", i, warc)
            body, err := GetWarc(ctx, s3Client, warc)
            if err != nil {
                fmt.Printf("error getting warc: %s", err)
                return
            }

            page, err := Parse(body)
            if err != nil {
                key := fmt.Sprintf("unparsed_warc/%s.warc", path)
                s3Client.PutObject(
                    ctx,
                    &s3.PutObjectInput{
                        Body:   bytes.NewReader(body),
                        Bucket: &s3Record.Bucket.Name,
                        Key:    &key,
                    },
                )
                fmt.Printf("error getting page %s: %s", key, err)
                return
            }

            mutex.Lock()
            err = pw.Write(page)
            pw.Flush(true)
            mutex.Unlock()
            if err != nil {
                fmt.Printf("error writing page: %s", err)
                return
            }
        })
    }

    limit.WaitAndClose()
    err = pw.WriteStop()
    if err != nil {
        return fmt.Errorf("error writing stop: %s", err)
    }
    bytesWriter.Flush()

    file.Seek(0, 0)
    _, err = s3Client.PutObject(
        ctx,
        &s3.PutObjectInput{
            Body:   file,
            Bucket: &s3Record.Bucket.Name,
            Key:    &key,
        },
    )
    if err != nil {
        return fmt.Errorf("error uploading warc: %s", err)
    }

    return nil
}

有没有办法避免将内容保存到临时文件中，并在编写器和上传函数之间仅使用有限大小的字节缓冲区？

换句话说，我可以开始将数据流式传输到读取器，同时仍然写入同一缓冲区吗？

正确答案

是的，有一种方法可以将相同的内容写入多个作者。使用 io.MultiWriter 可能允许您不使用临时文件。但是，使用临时文件可能仍然不错。

我经常使用 io.MultiWriter 写入校验和（sha256...）计算器列表。实际上，上次我阅读 S3 客户端代码时，我注意到它在幕后执行此操作来计算校验和。 MultiWriter 对于在云位置之间传输大文件非常有用。

此外，如果您最终使用临时文件。您可能想要使用 os.CreateTemp 创建临时文件。否则，如果您的代码在两个进程中运行或者您的文件具有相同的名称，则可能会遇到创建的文件名问题。

请随时澄清您的问题。我可以尝试再次回答:)

终于介绍完啦！小伙伴们，这篇关于《能否在 golang 中实现数据流式传输？》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！