go - Go 中的并行 zip 压缩

Question

我正在尝试从大量中小型文件构建一个 zip 存档。我希望能够同时执行此操作，因为压缩是 CPU 密集型的，而且我在多核服务器上运行。此外，我不想将整个存档保存在内存中，因为它可能会变得很大。

我的问题是，我是否必须压缩每个文件，然后手动合并所有内容以及 zip header 、校验和等？

如有任何帮助，我们将不胜感激。

Answer 1

我认为您不能组合 zip header 。

您可以做的是，运行 zip.Writer按顺序，在一个单独的 goroutine 中，然后为您要读取的每个文件生成一个新的 goroutine，并将它们通过管道传输到正在压缩它们的 goroutine。

这应该会减少您通过顺序读取文件获得的 IO 开销，尽管它可能不会利用多个内核来进行归档本身。

这是一个工作示例。请注意，为了简单起见，

• 它不能很好地处理错误，如果出现问题就会 panic ，
• 并且它没有过多地使用 defer 语句来演示事情发生的顺序。

由于延迟 is LIFO ，当您将它们堆叠在一起时，有时会让人感到困惑。

package main

import (
    "archive/zip"
    "io"
    "os"
    "sync"
)

func ZipWriter(files chan *os.File) *sync.WaitGroup {
    f, err := os.Create("out.zip")
    if err != nil {
        panic(err)
    }
    var wg sync.WaitGroup
    wg.Add(1)
    zw := zip.NewWriter(f)
    go func() {
        // Note the order (LIFO):
        defer wg.Done() // 2. signal that we're done
        defer f.Close() // 1. close the file
        var err error
        var fw io.Writer
        for f := range files {
            // Loop until channel is closed.
            if fw, err = zw.Create(f.Name()); err != nil {
                panic(err)
            }
            io.Copy(fw, f)
            if err = f.Close(); err != nil {
                panic(err)
            }
        }
        // The zip writer must be closed *before* f.Close() is called!
        if err = zw.Close(); err != nil {
            panic(err)
        }
    }()
    return &wg
}

func main() {
    files := make(chan *os.File)
    wait := ZipWriter(files)

    // Send all files to the zip writer.
    var wg sync.WaitGroup
    wg.Add(len(os.Args)-1)
    for i, name := range os.Args {
        if i == 0 {
            continue
        }
        // Read each file in parallel:
        go func(name string) {
            defer wg.Done()
            f, err := os.Open(name)
            if err != nil {
                panic(err)
            }
            files <- f
        }(name)
    }

    wg.Wait()
    // Once we're done sending the files, we can close the channel.
    close(files)
    // This will cause ZipWriter to break out of the loop, close the file,
    // and unblock the next mutex:
    wait.Wait()
}

用法:go run example.go/path/to/*.log。

这是事情发生的顺序:

1. 打开输出文件进行写入。
2. 创建 zip.Writer使用该文件。
3. 启动 goroutine 监听 channel 上的文件。
4. 遍历每个文件，这可以在每个文件的一个 goroutine 中完成。
5. 将每个文件发送到第 3 步中创建的 goroutine。
6. 在 goroutine 中处理完每个文件后，关闭文件以释放资源。
7. 将每个文件发送到所述 goroutine 后，关闭 channel 。
8. 等到压缩完成(按顺序完成)。
9. 一旦压缩完成( channel 耗尽)，zip writer 应该关闭。
10. 只有当 zip writer 关闭时，输出文件才应该关闭。
11. 最后一切都关闭了，所以关闭 sync.WaitGroup 以告诉调用函数我们可以开始了。 (这里也可以使用 channel ，但是 sync.WaitGroup 看起来更优雅。)
12. 当您从 zip writer 收到一切已正确关闭的信号时，您可以从 main 退出并正常终止。

这可能无法回答您的问题，但我之前一直在使用类似的代码为 Web 服务即时生成 zip 存档。它表现得非常好，即使实际的压缩是在一个 goroutine 中完成的。克服 IO 瓶颈已经是一种进步。