Golang并发，批量处理item

Question

我正在编写一个程序来处理文本文件中的数百万行，500k 需要 5 秒来验证文件，我想加快速度。

我想循环遍历这些项目并异步处理其中的 x 个，然后等待响应以查看是否应该继续。

我写了一些虚拟代码，我不确定我写的内容是否有意义，只是看起来相当复杂，有没有更简单更优雅的方法。

package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    // Need an object to loop over
    // need a loop to read the response
    items := 100000
    concurrency := 20
    sem := make(chan bool, concurrency)
    returnChan := make(chan error)
    finChan := make(chan bool)

    var wg sync.WaitGroup

    go func() {
        for x := 0; x < items; x++ {
            // loop over all items
            // only do maxitems at a time
            wg.Add(1)
            sem <- true
            go delayFunc(x, sem, returnChan, &wg)
        }
        wg.Wait()
        finChan <- true
    }()

    var err error
    finished := false
    for {
        select {
        case err = <-returnChan:
            if err != nil {
                break
            }
        case _ = <-finChan:
            finished = true
            break
        default:
            continue
        }

        if err != nil || finished == true {
            break
        }
    }
    fmt.Println(err)
}

func delayFunc(x int, sem chan bool, returnChan chan error, wg *sync.WaitGroup) {
    //fmt.Printf("PROCESSING (%v)\n", x)
    time.Sleep(10 * time.Millisecond)
    <-sem // release the lock
    wg.Done()
    if x == 95000 {
        returnChan <- fmt.Errorf("Something not right")
    } else {
        returnChan <- nil
    }
}

Answer 1

您的代码看起来不错，您实现了 Go 中常用的模式。缺点是 - 你为每个项目生成工人 goroutine。廉价生成 goroutine 并不是免费的。另一种方法是产生 N 个工人并通过渠道为他们提供物品。像这样的

package main
import (
    "fmt"
    "time"
)

func main() {
    items := 100
    concurrency := 10
    in := make(chan int)
    ret := make(chan error)

    for x := 0; x < concurrency; x++ {
        go worker(in, ret)
    }
    go func() {
        for x := 0; x < items; x++ {
            // loop over all items
            in <- x
        }
        close(in)
    }()
    for err := range ret {
        if err != nil {
            fmt.Println(err.Error())
            break
        }
    }
}
func worker(in chan int, returnChan chan error) {
    //fmt.Printf("PROCESSING (%v)\n", x)
    for x := range in {
        if x == 95 {
            returnChan <- fmt.Errorf("Something not right")
        } else {
            returnChan <- nil
        }
        time.Sleep(10 * time.Millisecond)
    }
    returnChan <- fmt.Errorf("The End")
}

Playground