GoLang: Decompress bz2 in on goroutine, потребляйте в другом goroutine

Question

Я изучаю SWE в новом граде Go (и люблю его).

Я создаю парсер для файлов дампа Wikipedia - в основном огромный XML-файл с сжатым файлом bzip2 (~ 50 ГБ без сжатия).

Я хочу сделать как потоковое декомпрессии, так и разбор, что звучит достаточно просто. Для декомпрессии, я:

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

А затем передать читателя к XML-парсер:

decoder := xml.NewDecoder(inputFile)

Однако, так как декомпрессия и синтаксический дорогие операции, я хотел бы иметь они запускаются на отдельных процедурах Go, чтобы использовать дополнительные ядра. Как я буду делать это в Go?

Единственное, что я могу придумать, это обернуть файл в chan [] байт и реализовать интерфейс io.Reader, но я полагаю, что это может быть способ создания (и более чистого) использования.

Кто-нибудь когда-либо делал что-то подобное?

Спасибо! Manuel

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