xml - GoLang : Decompress bz2 in on goroutine, 在其他协程中消费

Question

我是一名刚毕业的 SWE，正在学习 Go(并且喜欢它)。

我正在为维基百科转储文件构建一个解析器——基本上是一个巨大的 bzip2 压缩 XML 文件(~50GB 未压缩)。

我想同时做流式解压和解析，听起来很简单。对于减压，我这样做:

inputFilePath := flag.Arg(0)
输入阅读器 := bzip2.NewReader(输入文件)

然后将读取器传递给 XML 解析器:

解码器 := xml.NewDecoder(inputFile)

但是，由于解压缩和解析都是昂贵的操作，我想让它们在单独的 Go 例程上运行以利用额外的核心。我将如何在 Go 中执行此操作？

我唯一能想到的是将文件包装在一个 chan []byte 中，并实现 io.Reader 接口(interface)，但我认为可能有一种内置的方式(和更清晰的)来实现它。

有没有人做过这样的事情？

谢谢!曼纽尔

Answer 1

您可以使用 io.Pipe , 然后使用 io.Copy将解压后的数据推送到管道中，并在另一个 goroutine 中读取:

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "io"
    "sync"
)

func main() {

    rawJson := []byte(`{
            "Foo": {
                "Bar": "Baz"
            }
        }`)

    bzip2Reader := bytes.NewReader(rawJson) // this stands in for the bzip2.NewReader

    var wg sync.WaitGroup
    wg.Add(2)

    r, w := io.Pipe()

    go func() {
        // write everything into the pipe. Decompression happens in this goroutine.
        io.Copy(w, bzip2Reader)
        w.Close()
        wg.Done()
    }()

    decoder := json.NewDecoder(r)

    go func() {
        for {
            t, err := decoder.Token()
            if err != nil {
                break
            }
            fmt.Println(t)
        }
        wg.Done()
    }()

    wg.Wait()
}

http://play.golang.org/p/fXLnfnaWYA