go - 我想将一个文件拆分为大小相同的 "chunks"或 slice 并使用 goroutines 同时处理它们

Question

使用 Go，我有很大的日志文件。目前我打开它们，创建一个新的扫描器 bufio.NewScanner，然后 for scanner.Scan() 循环遍历这些行。每行都通过一个处理函数发送，该函数将其与正则表达式匹配并提取数据。我想使用 goroutines 同时以 block 的形式处理这个文件。我相信这可能比按顺序遍历整个文件更快。

每个文件可能需要几秒钟，我想知道我是否可以一次处理一个文件，比如 10 个文件。我相信如果需要我可以牺牲内存。我有 ~3gb，最大的日志文件可能有 75mb。

我看到 scanner 有一个 .Split() 方法，您可以在其中提供自定义拆分功能，但我找不到好的解决方案使用这种方法。

我还尝试创建一个 slice 的 slice ，使用 scanner.Scan() 循环遍历扫描仪并将 scanner.Text() 附加到每个 slice 。例如:

// pseudocode because I couldn't get this to work either

scanner := bufio.NewScanner(logInfo)
threads := [[], [], [], [], []]

i := 0
for scanner.Scan() {
    i = i + 1
    if i > 5 {
        i = 0
    }
    threads[i] = append(threads[i], scanner.Text())
}
fmt.Println(threads)

我是 Go 的新手，很关心效率和性能。我想学习如何编写好的 Go 代码!非常感谢任何帮助或建议。

Answer 1

Peter 提供了一个很好的起点，如果你想做类似扇出、扇入模式的事情，你可以这样做:

package main

import (
    "bufio"
    "fmt"
    "log"
    "os"
    "sync"
)

func main() {
    file, err := os.Open("/path/to/file.txt")
    if err != nil {
        log.Fatal(err)
    }
    defer file.Close()

    lines := make(chan string)
    // start four workers to do the heavy lifting
    wc1 := startWorker(lines)
    wc2 := startWorker(lines)
    wc3 := startWorker(lines)
    wc4 := startWorker(lines)
    scanner := bufio.NewScanner(file)
    go func() {
        defer close(lines)
        for scanner.Scan() {
            lines <- scanner.Text()
        }

        if err := scanner.Err(); err != nil {
            log.Fatal(err)
        }
    }()

    merged := merge(wc1, wc2, wc3, wc4)
    for line := range merged {
        fmt.Println(line)
    }
}

func startWorker(lines <-chan string) <-chan string {
    finished := make(chan string)
    go func() {
        defer close(finished)
        for line := range lines {
            // Do your heavy work here
            finished <- line
        }
    }()
    return finished
}

func merge(cs ...<-chan string) <-chan string {
    var wg sync.WaitGroup
    out := make(chan string)

    // Start an output goroutine for each input channel in cs.  output
    // copies values from c to out until c is closed, then calls wg.Done.
    output := func(c <-chan string) {
        for n := range c {
            out <- n
        }
        wg.Done()
    }
    wg.Add(len(cs))
    for _, c := range cs {
        go output(c)
    }

    // Start a goroutine to close out once all the output goroutines are
    // done.  This must start after the wg.Add call.
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}