如何使用通道收集来自各种 goroutine 的响应

Question

我是 Golang 的新手，我有一个使用 实现的任务，我WaitGroup想Mutex将其转换为使用Channels。该任务的一个非常简短的描述是：根据需要拒绝尽可能多的 go 例程来处理结果，并在主 go 例程中等待并收集所有结果。我使用的实现WaitGroup如下Mutex：package mainimport (    "fmt"    "math/rand"    "sync"    "time")func process(input int, wg *sync.WaitGroup, result *[]int, lock *sync.Mutex) *[]int {    defer wg.Done()    defer lock.Unlock()    rand.Seed(time.Now().UnixNano())    n := rand.Intn(5)    time.Sleep(time.Duration(n) * time.Second)    lock.Lock()    *result = append(*result, input * 10)    return result}func main() {    var wg sync.WaitGroup    var result []int    var lock sync.Mutex    for i := range []int{1,2,3,4,5} {        wg.Add(1)        go process(i, &wg, &result, &lock)    }}如何使用Mutexto替换内存同步Channels？我的主要问题是我不确定如何确定处理最终任务的最终 go 例程，因此让那个成为关闭channel. 这个想法是，通过关闭channel主 go 例程可以循环遍历channel，检索结果，当它看到channel已经关闭时，它继续前进。在这种情况下，关闭通道的方法也可能是错误的，因此我在这里问。更有经验的 go 程序员如何使用 解决这个问题channels？

Answer 1

这是一个使用WaitGroup而不是等待固定数量的结果的解决方案。

package main

import (

    "fmt"

    "math/rand"

    "sync"

    "time"

)

func process(input int, wg *sync.WaitGroup, resultChan chan<- int) {

    defer wg.Done()

    rand.Seed(time.Now().UnixNano())

    n := rand.Intn(5)

    time.Sleep(time.Duration(n) * time.Second)

    resultChan <- input * 10

}

func main() {

    var wg sync.WaitGroup

    resultChan := make(chan int)

    for i := range []int{1,2,3,4,5} {

        wg.Add(1)

        go process(i, &wg, resultChan)

    }

    go func() {

        wg.Wait()

        close(resultChan)

    }()

    var result []int

    for r := range resultChan {

        result = append(result, r)

    }

    fmt.Println(result)

}

Answer 2

我更改了您的代码以使用该频道。还有许多其他方法可以使用该频道。

package main

import (

    "fmt"

    "math/rand"

    "time"

)

func process(input int, out chan<- int) {

    rand.Seed(time.Now().UnixNano())

    n := rand.Intn(5)

    time.Sleep(time.Duration(n) * time.Second)

    out <- input * 10

}

func main() {

    var result []int

    resultChan := make(chan int)

    items := []int{1, 2, 3, 4, 5}

    for _, v := range items {

        go process(v, resultChan)

    }

    for i := 0; i < len(items); i++ {

        res, _ := <-resultChan

        result = append(result, res)

    }

    close(resultChan)

    fmt.Println(result)

}

更新：（评论的答案）

如果项目数未知，则需要向主发出信号以完成。否则“死锁”，您可以创建一个通道来指示主要功能完成。此外，您可以使用sync.waiteGroup.

对于 Goroutine 中的 panic，你可以使用 defer 和 recover 来处理错误。并且您可以创建一个可以使用的错误通道矿石x/sync/errgroup。

有很多解决方案。这取决于你的问题。所以没有具体的方式来使用 goroutine、channel 和...

Answer 3

这是一个示例片段，其中我使用通道切片而不是等待组来执行分叉连接：

package main

import (

    "fmt"

    "os"

)

type cStruct struct {

    resultChan chan int

    errChan    chan error

}

func process(i int) (v int, err error) {

    v = i

    return

}

func spawn(i int) cStruct {

    r := make(chan int)

    e := make(chan error)

    go func(i int) {

        defer close(r)

        defer close(e)

        v, err := process(i)

        if err != nil {

            e <- err

            return

        }

        r <- v

        return

    }(i)

    return cStruct{

        r,

        e,

    }

}

func main() {

    //have a slice of channelStruct

    var cStructs []cStruct

    nums := []int{1, 2, 3, 4, 5}

    for _, v := range nums {

        cStruct := spawn(v)

        cStructs = append(cStructs, cStruct)

    }

    //All the routines have been spawned, now iterate over the slice:

    var results []int

    for _, c := range cStructs {

        rChan, errChan := c.resultChan, c.errChan

        select {

        case r := <-rChan:

            {

                results = append(results, r)

            }

        case err := <-errChan:

            {

                if err != nil {

                    os.Exit(1)

                    return

                }

            }

        }

    }

    //All the work should be done by now, iterating over the results

    for _, result := range results {

        fmt.Println("Aggregated result:", result)

    }

}