如何高效地并行化数组列表并控制并行度？

在IT行业这个发展更新速度很快的行业，只有不停止的学习，才不会被行业所淘汰。如果你是Golang学习者，那么本文《如何高效地并行化数组列表并控制并行度？》就很适合你！本篇内容主要包括##content_title##，希望对大家的知识积累有所帮助，助力实战开发！

我有一个 resourceid 数组，我需要并行循环。并为每个资源生成url，然后放入一个map中，其中key（resourcid）是，value是url。

我得到了下面的代码，它可以完成这项工作，但我不确定这是否是正确的方法。我在这里使用 sizedwaitgroup 来并行化 resourceid 列表。并且在向地图写入数据时也使用地图锁。我确信这不是有效的代码，因为使用锁然后使用调整大小的等待组会产生一些性能问题。

最好、最有效的方法是什么？我应该在这里使用频道吗？我想控制应该有多少并行度，而不是 resourceid 列表的运行长度。如果任何 resourceid url 生成失败，我想将其记录为该 resourceid 的错误，但不要中断其他并行运行的 go 例程以获取为其他 resourceid 生成的 url。

例如：如果有 10 个资源，其中 2 个失败，则记录这 2 个资源的错误，并且地图应该包含剩余 8 个资源的条目。

// running 20 threads in parallel
swg := sizedwaitgroup.New(20)
var mutex = &sync.Mutex{}
start := time.Now()
m := make(map[string]*customerPbV1.CustomerResponse)
for _, resources := range resourcesList {
  swg.Add()
  go func(resources string) {
    defer swg.Done()
    customerUrl, err := us.GenerateUrl(clientId, resources, appConfig)
    if err != nil {
      errs.NewWithCausef(err, "Could not generate the url for %s", resources)
    }
    mutex.Lock()
    m[resources] = customerUrl
    mutex.Unlock()
  }(resources)
}
swg.Wait()

elapsed := time.Since(start)
fmt.Println(elapsed)

注意：以上代码将从多个读取器线程以高吞吐量调用，因此需要良好的性能。

正确答案

我不确定 sizedwaitgroup 是什么，也没有解释，但总的来说，这种方法看起来不是很典型的 go。就此而言，“最好”是一个见仁见智的问题，但 go 中最典型的方法是这样的：

func main() {
    wg := new(sync.waitgroup)
    start := time.now()
    numworkers := 20
    m := make(map[string]*customerpbv1.customerresponse)
    work := make(chan string)
    results := make(chan result)
    for i := 0; i < numworkers; i++ {
        wg.add(1)
        go worker(work, results)
    }
    go func() {
        for _, resources := range resourceslist {
            work <- resources
        }
        close(work)
    }()

    go func() {
        wg.wait()
        close(results)
    }()

    for result := range results {
        m[result.resources] = result.response
    }

    elapsed := time.since(start)
    fmt.println(elapsed)
}

type result struct {
    resources string
    response  *customerpbv1.customerresponse
}

func worker(ch chan string, r chan result) {
    for w := range ch {
        customerurl, err := us.generateurl(clientid, w, appconfig)
        if err != nil {
            errs.newwithcausef(err, "could not generate the url for %s", resources)
            continue
        }
        r <- result{w, customerurl}
    }
}

（不过，根据名称，我假设 errs.newwithcause 实际上并不处理错误，而是返回一个错误，在这种情况下，当前代码会将它们扔在地板上，正确的解决方案将有一个额外的 chan error 用于处理错误：

func main() {
    wg := new(sync.waitgroup)
    start := time.now()
    numworkers := 20
    m := make(map[string]*customerpbv1.customerresponse)
    work := make(chan string)
    results := make(chan result)
    errors := make(chan error)
    for i := 0; i < numworkers; i++ {
        wg.add(1)
        go worker(work, results, errors)
    }

    go func() {
        for _, resources := range resourceslist {
            work <- resources
        }
        close(work)
    }()

    go func() {
        wg.wait()
        close(results)
        close(errors)
    }()

    go func() {
        for err := range errors {
            // do something with err
        }
    }()

    for result := range results {
        m[result.resources] = result.response
    }

    elapsed := time.since(start)
    fmt.println(elapsed)
}

type result struct {
    resources string
    response  *customerpbv1.customerresponse
}

func worker(ch chan string, r chan result, errs chan error) {
    for w := range ch {
        customerurl, err := us.generateurl(clientid, w, appconfig)
        if err != nil {
            errs <- errs.newwithcausef(err, "could not generate the url for %s", resources)
            continue
        }
        r <- result{w, customerurl}
    }
}

我已经创建了带有注释的示例代码。 请阅读评论。

package main

import (
    "errors"
    "fmt"
    "log"
    "math/rand"
    "runtime"
    "strconv"
    "sync"
    "time"
)

type Result struct {
    resource string
    val      int
    err      error
}

/*
CHANGE Result struct to this
result struct will collect all you need to create map
type Result struct {
    resources string
    customerUrl *customerPbV1.CustomerResponse
    err error
}
*/

// const numWorker = 8

func main() {
    now := time.Now()
    rand.Seed(time.Now().UnixNano())
    m := make(map[string]int)
    // m := make(map[string]*customerPbV1.CustomerResponse)                 // CHANGE TO THIS

    numWorker := runtime.NumCPU()
    fmt.Println(numWorker)
    chanResult := make(chan Result)

    go func() {
        for i := 0; i < 20; i++ {
            /*
             customerUrl, err := us.GenerateUrl(clientId, resources, appConfig)
             we asume i is resources
             chanResult <- Result {resource: strconv.Itoa(i)}
            */
            chanResult <- Result{ // this will block until chanResult is consume in line 68
                resource: strconv.Itoa(i),
            }
        }
        close(chanResult)
    }()

    var wg sync.WaitGroup
    cr := make(chan Result)
    wg.Add(numWorker)

    go func() {
        wg.Wait()
        close(cr) // NOTE: don't forget to close cr
    }()

    go func() {
        for i := 0; i < numWorker; i++ { // this for loop will run goroutine
            go func(x int) {
                for job := range chanResult { // unblock chan on line 49
                    log.Println("worker", x, "working on", job.resource)
                    x, err := query(job.resource) // TODO: customerUrl, err := us.GenerateUrl(clientId, resources, appConfig)
                    cr <- Result{                 // send to channel, will block until it consume. Consume is in MAIN goroutine "line 84"
                        resource: job.resource,
                        val:      x,
                        err:      err,
                    }
                }
                wg.Done()
            }(i)
        }
    }()

    counterTotal := 0
    counterSuccess := 0
    for res := range cr { // will unblock channel in line 71
        if res.err != nil {
            log.Printf("error found %s. stack trace: %s", res.resource, res.err)
        } else {
            m[res.resource] = res.val // NOTE: save to map
            counterSuccess++
        }
        counterTotal++
    }
    log.Printf("%d/%d of total job run", counterSuccess, counterTotal)
    fmt.Println("final :", m)
    fmt.Println("len m", len(m))

    fmt.Println(runtime.NumGoroutine())
    fmt.Println(time.Since(now))
}

func query(s string) (int, error) {
    time.Sleep(time.Second)
    i, err := strconv.Atoi(s)
    if err != nil {
        return 0, err
    }

    if i%3 == 0 {
        return 0, errors.New("i divided by 3")
    }
    ms := i + 500 + rand.Intn(500)
    return ms, nil
}

演示：https://go.dev/play/p/LeyE9n1hh81

到这里，我们也就讲完了《如何高效地并行化数组列表并控制并行度？》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！