GoLang: Dekomprimieren bz2 in Goroutine, konsumieren in anderen Goroutine

Question

Ich bin ein New-grad SWE Lernen Go (und liebe es).

Ich baue einen Parser für Wikipedia-Dump-Dateien - im Grunde eine riesige bzip2-komprimierte XML-Datei (~ 50GB unkomprimiert).

Ich möchte sowohl Streaming-Dekomprimierung und Parsing, die einfach genug klingt. Für Dekompression, das tue ich:

inputFilePath := flag.Arg(0) inputReader := bzip2.NewReader(inputFile)

Und dann den Leser auf den XML-Parser übergeben:

decoder := xml.NewDecoder(inputFile)

Da jedoch beide Dekomprimieren und Parsing teure Operationen sind, würde Ich mag haben Sie werden in separaten Go-Routinen ausgeführt, um zusätzliche Kerne zu verwenden. Wie würde ich das in Go machen?

Das einzige, was ich mir vorstellen kann, ist das Wrappen der Datei in einem chan [] Byte und das Implementieren der io.Reader-Schnittstelle, aber ich nehme an, dass es einen gebauten (und saubereren) Weg gibt.

Hat schon mal jemand so etwas gemacht?

Danke! Manuel

Answer 1

Sie können io.Pipe verwenden, dann io.Copy verwenden, um die dekomprimierten Daten in das Rohr zu schieben, und in einem anderen goroutine lesen:

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