从管道读取Golang读取大量数据

Question

我正在尝试读取一个被tar,流式传输到stdin的存档,但我在某种程度上读取的数据远远超过tar发送的数据.

我像这样运行我的命令:

tar -cf - somefolder | ./my-go-binary

源代码是这样的:

package main

import (
    "bufio"
    "io"
    "log"
    "os"
)

// Read from standard input
func main() {
    reader := bufio.NewReader(os.Stdin)
    // Read all data from stdin, processing subsequent reads as chunks.
    parts := 0
    for {
        parts++
        data := make([]byte, 4<<20) // Read 4MB at a time
        _, err := reader.Read(data)
        if err == io.EOF {
            break
        } else if err != nil {
            log.Fatalf("Problems reading from input: %s", err)
        }
    }
    log.Printf("Total parts processed: %d\n", parts)
}

对于一个100MB的tarred文件夹,我得到1468块4MB(即6.15GB)!此外,data []byte阵列的大小似乎并不重要:如果我将块大小设置为40MB,我仍然可以得到~1400块40MB的数据,这根本没有意义.

有什么我需要做的事情来os.Stdin正确地使用Go 读取数据吗？

Answer 1

你的代码效率低下.它data通过循环每次分配和初始化.

for {
    data := make([]byte, 4<<20) // Read 4MB at a time
}

为你的代码reader作为io.Reader是错误的.例如,您忽略读取的字节数,_, err := reader.Read(data)并且不能err正确处理错误.

包io

import "io" 

类型读者

type Reader interface {
        Read(p []byte) (n int, err error)
}

Reader是包装基本Read方法的接口.

读取读取最多len(p)个字节到p.它返回读取的字节数(0 <= n <= len(p))和遇到的任何错误.即使Read返回n <len(p),它也可以在调用期间将所有p用作临时空间.如果某些数据可用但不是len(p)字节,则Read通常返回可用而不是等待更多的数据.

当Read在成功读取n> 0字节后遇到错误或文件结束条件时,它返回读取的字节数.它可以从同一个调用返回(非零)错误,或者从后续调用中返回错误(和n == 0).这种一般情况的一个例子是Reader在输入流的末尾返回非零数量的字节可能返回err == EOF或err == nil.下一个Read应该返回0,EOF无论如何.

在考虑错误错误之前,调用者应始终处理返回的n> 0个字节.这样做可以正确处理读取一些字节后发生的I/O错误以及两种允许的EOF行为.

除非len(p)== 0,否则不鼓励读取的实现返回零字节计数.调用者应该将0和nil的返回视为表示没有发生任何事情; 特别是它并不表示EOF.

实现不得保留p.

这是一个符合io.Reader界面的模型文件读取程序:

package main

import (
    "bufio"
    "io"
    "log"
    "os"
)

func main() {
    nBytes, nChunks := int64(0), int64(0)
    r := bufio.NewReader(os.Stdin)
    buf := make([]byte, 0, 4*1024)
    for {
        n, err := r.Read(buf[:cap(buf)])
        buf = buf[:n]
        if n == 0 {
            if err == nil {
                continue
            }
            if err == io.EOF {
                break
            }
            log.Fatal(err)
        }
        nChunks++
        nBytes += int64(len(buf))
        // process buf
        if err != nil && err != io.EOF {
            log.Fatal(err)
        }
    }
    log.Println("Bytes:", nBytes, "Chunks:", nChunks)
}

输出:

2014/11/29 10:00:05 Bytes: 5589891 Chunks: 1365

Answer 2

阅读Read的文档:

读取将数据读入p.它返回读入p的字节数.它在底层Reader上最多调用一次,因此n可能小于len(p).在EOF,计数将为零,错误将为io.EOF.

你一次不读4MB.您正在提供缓冲区空间并丢弃可能告诉您Read实际读取的数量的整数.缓冲区空间是最大的,但大多数情况下128k似乎每次调用都会被读取,至少在我的系统上是这样.亲自尝试一下:

// Read from standard input
func main() {
    reader := bufio.NewReader(os.Stdin)
    // Read all data from stdin, passing the data as parts into the channel
    // for processing.
    parts := 0
    for {
        parts++
        data := make([]byte, 4<<20) // Read 4MB at a time
        amount , err := reader.Read(data)
        // WILL NOT BE 4MB!
        log.Printf("Read: %v\n", amount)
        if err == io.EOF {
            break
        } else if err != nil {
            log.Fatalf("Problems reading from input: %s", err)
        }
    }
    log.Printf("Total parts processed: %d\n", parts)
}

您必须实现处理不同读取量的逻辑.