限制 goroutine 处理大型 csv 文件

在处理大型 CSV 文件时，使用工作池可以有效地并发执行任务。然而，创建过多 goroutine 可能会耗尽内存，导致程序崩溃。为了限制 goroutine 数量，可以采用以下方法： 首先，根据系统资源和任务需求确定一个合理的 goroutine 数目。其次，使用信道和协程同步机制来控制 goroutine 的创建和销毁。通过在信道上发送信号或使用上下文取消机制，可以优雅地终止 goroutine。此外，可以通过使用锁或其他同步原语来限制对共享资源的并发访问，避免竞争条件。

我正在尝试找到读取 csv 文件（约 1m 行）的最有效方法。 每行都包含一个指向我需要下载的图像的 http 链接。

这是我当前使用工作池的代码：

func worker(queue chan []string, worknumber int, done, ks chan bool) {
    for true {
        select {
        case url := <-queue:
            fmt.Println("doing work!", url, "worknumber", worknumber)
            processData(url) // HTTP download
            done <- true
        case <-ks:
            fmt.Println("worker halted, number", worknumber)
            return
        }
    }
}

func main() {
    start := time.Now()
    flag.Parse()
    fmt.Print(strings.Join(flag.Args(), "\n"))
    if *filename == "REQUIRED" {
        return
    }

    csvfile, err := os.Open(*filename)
    if err != nil {
        fmt.Println(err)
        return
    }
    count, _ := lineCounter(csvfile)
    fmt.Printf("Total count: %d\n", count)
    csvfile.Seek(0, 0)

    defer csvfile.Close()

    //bar := pb.StartNew(count)
    bar := progressbar.NewOptions(count)
    bar.RenderBlank()

    reader := csv.NewReader(csvfile)

    //channel for terminating the workers
    killsignal := make(chan bool)

    //queue of jobs
    q := make(chan []string)
    // done channel takes the result of the job
    done := make(chan bool)

    numberOfWorkers := *numChannels
    for i := 0; i < numberOfWorkers; i++ {
        go worker(q, i, done, killsignal)
    }

    i := 0
    for {
        record, err := reader.Read()
        if err == io.EOF {
            break
        } else if err != nil {
            fmt.Println(err)
            return
        }
        i++

        go func(r []string, i int) {
            q <- r
            bar.Add(1)
        }(record, i)
    }

    // a deadlock occurs if c >= numberOfJobs
    for c := 0; c < count; c++ {
        <-done
    }

    fmt.Println("finished")

    // cleaning workers
    close(killsignal)
    time.Sleep(2 * time.Second)

    fmt.Printf("\n%2fs", time.Since(start).Seconds())
}

我的问题是它打开了很多 goroutine，使用了所有内存并崩溃。

限制它的最佳方法是什么？

解决方案

我去掉了进度条，因为我不想打扰它，但总的来说，这更接近您正在寻找的内容。

它并没有真正处理错误，它们只是在致命状态下失败。

我添加了上下文和取消支持。

您可能想要检查 https://godoc.org/golang.org/x/sync/errgroup#Group.Go

作为一般建议，您需要学习 golang 模式及其用法。

很明显你还做得不够，或者你正在学习中。

它根本不是最快的程序，但它可以完成任务。

这只是一个草稿，旨在让您回到更好的方向。

package main

import (
    "context"
    "encoding/csv"
    "flag"
    "fmt"
    "io"
    "log"
    "os"
    "os/signal"
    "sync"
    "time"
)

func worker(ctx context.context, dst chan string, src chan []string) {
    for {
        select {
        case url, ok := <-src: // you must check for readable state of the channel.
            if !ok {
                return
            }
            dst <- fmt.sprintf("out of %v", url) // do somethingg useful.
        case <-ctx.done(): // if the context is cancelled, quit.
            return
        }
    }
}

func main() {

    // create a context
    ctx, cancel := context.withcancel(context.background())
    defer cancel()
    // that cancels at ctrl+c
    go onsignal(os.interrupt, cancel)

    // parse command line arguments
    var filename string
    var numberofworkers int
    flag.stringvar(&filename, "filename", "", "src file")
    flag.intvar(&numberofworkers, "c", 2, "concurrent workers")
    flag.parse()

    // check arguments
    if filename == "" {
        log.fatal("filename required")
    }

    start := time.now()

    csvfile, err := os.open(filename)
    if err != nil {
        log.fatal(err)
    }
    defer csvfile.close()

    reader := csv.newreader(csvfile)

    // create the pair of input/output channels for the controller=>workers com.
    src := make(chan []string)
    out := make(chan string)

    // use a waitgroup to manage synchronization
    var wg sync.waitgroup

    // declare the workers
    for i := 0; i < numberofworkers; i++ {
        wg.add(1)
        go func() {
            defer wg.done()
            worker(ctx, out, src)
        }()
    }

    // read the csv and write it to src
    go func() {
        for {
            record, err := reader.read()
            if err == io.eof {
                break
            } else if err != nil {
                log.fatal(err)
            }
            src <- record // you might select on ctx.done().
        }
        close(src) // close src to signal workers that no more job are incoming.
    }()

    // wait for worker group to finish and close out
    go func() {
        wg.wait() // wait for writers to quit.
        close(out) // when you close(out) it breaks the below loop.
    }()

    // drain the output
    for res := range out {
        fmt.println(res)
    }

    fmt.printf("\n%2fs", time.since(start).seconds())
}

func onsignal(s os.signal, h func()) {
    c := make(chan os.signal, 1)
    signal.notify(c, s)
    <-c
    h()
}

您正在为文件中的每一行创建一个新的 goroutine。这就是为什么。如果您已经拥有所需的工作人员，则没有理由这样做。

简而言之，改变这个：

go func(r []string, i int) {
        q <- r
        bar.add(1)
    }(record, i)

对此：

q <- record
    bar.Add(1)

今天关于《限制 goroutine 处理大型 csv 文件》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于的内容请关注golang学习网公众号！