goroutine 等待频道的响应并继续

Question

我正在学习并发，我想实现一个简单的示例，该示例从矩阵中获取行并将值的数组（切片）添加到每一行。

由于我使用的是通道，因此我尝试等待每一行从 goroutine 获得相应的结果。但是，这并不比仅同步执行此操作更好。如何让每一行等待它们各自的结果并允许其他行同时计算它们的结果？

https://play.golang.org/p/uCOGwOBeIQL

package main

import "fmt"

/*

Array:

0 1 2 3 4 5 6 7 8 9

+

Matrix:

1 0 0 0 0 0 0 0 0 0

0 1 0 0 0 0 0 0 0 0

0 0 1 0 0 0 0 0 0 0

0 0 0 1 0 0 0 0 0 0

0 0 0 0 1 0 0 0 0 0

0 0 0 0 0 1 0 0 0 0

0 0 0 0 0 0 1 0 0 0

0 0 0 0 0 0 0 1 0 0

0 0 0 0 0 0 0 0 1 0

0 0 0 0 0 0 0 0 0 1

-> 

Expected result:

1 1 2 3 4 5 6 7 8 9

0 2 2 3 4 5 6 7 8 9

0 1 3 3 4 5 6 7 8 9

0 1 2 4 4 5 6 7 8 9

0 1 2 3 5 5 6 7 8 9

0 1 2 3 4 6 6 7 8 9

0 1 2 3 4 5 7 7 8 9

0 1 2 3 4 5 6 8 8 9

0 1 2 3 4 5 6 7 9 9

0 1 2 3 4 5 6 7 8 10

*/

func main() {

    numbers := []int {0,1,2,3,4,5,6,7,8,9}

    matrix := [][]int{

        {1,0,0,0,0,0,0,0,0,0},

        {0,1,0,0,0,0,0,0,0,0},

        {0,0,1,0,0,0,0,0,0,0},

        {0,0,0,1,0,0,0,0,0,0},

        {0,0,0,0,1,0,0,0,0,0},

        {0,0,0,0,0,1,0,0,0,0},

        {0,0,0,0,0,0,1,0,0,0},

        {0,0,0,0,0,0,0,1,0,0},

        {0,0,0,0,0,0,0,0,1,0},

        {0,0,0,0,0,0,0,0,0,1},

    }

    rmatrix := make([][]int, 10)

    for i, row := range matrix {

        cResult := make(chan []int)

        go func(row []int, numbers []int, c chan <- []int) {

            c <- addRow(row,numbers)

        }(row,numbers,cResult)

        //this read from the channel will block until the goroutine sends its result over the channel

        rmatrix[i] = <- cResult

    }

    fmt.Println(rmatrix)

}

func addRow(row []int, numbers []int) []int{

    result := make([]int, len(row))

    for i,e := range row {

        result[i] = e + numbers[i];

    }

    return result

}

Answer 1

我需要使用 async.WaitGroup并直接分配调用的结果（以保证它们返回到索引行）。谢谢@彼得package mainimport (&nbsp; &nbsp; "fmt"&nbsp; &nbsp; "sync")func main() {&nbsp; &nbsp; numbers := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}&nbsp; &nbsp; matrix := [][]int{&nbsp; &nbsp; &nbsp; &nbsp; {1, 0, 0, 0, 0, 0, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 1, 0, 0, 0, 0, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 1, 0, 0, 0, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 1, 0, 0, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 1, 0, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 0, 1, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 0, 0, 1, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 0, 0, 0, 1, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 0, 0, 0, 0, 1, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 0, 0, 0, 0, 0, 1},&nbsp; &nbsp; }&nbsp; &nbsp; rmatrix := make([][]int, 10)&nbsp; &nbsp; var waitGroup sync.WaitGroup&nbsp; &nbsp; for i, row := range matrix {&nbsp; &nbsp; &nbsp; &nbsp; waitGroup.Add(1)&nbsp; &nbsp; &nbsp; &nbsp; go func(i int, row []int) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; rmatrix[i] = addRow(row, numbers)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; waitGroup.Done()&nbsp; &nbsp; &nbsp; &nbsp; }(i, row)&nbsp; &nbsp; }&nbsp; &nbsp; waitGroup.Wait()&nbsp; &nbsp; fmt.Println(rmatrix)}func addRow(row []int, numbers []int) []int {&nbsp; &nbsp; result := make([]int, len(row))&nbsp; &nbsp; for i, e := range row {&nbsp; &nbsp; &nbsp; &nbsp; result[i] = e + numbers[i]&nbsp; &nbsp; }&nbsp; &nbsp; return result}

Answer 2

这个例子产生了较少数量的 goroutine，并且也保证了正确的顺序，不管哪个 goroutine 先完成了它的处理。package mainimport (&nbsp; &nbsp; "fmt"&nbsp; &nbsp; "sync")type rowRes struct {&nbsp; &nbsp; index&nbsp; int&nbsp; &nbsp; result *[]int}func addRow(index int, row []int, numbers []int) rowRes {&nbsp; &nbsp; result := make([]int, len(row))&nbsp; &nbsp; for i, e := range row {&nbsp; &nbsp; &nbsp; &nbsp; result[i] = e + numbers[i]&nbsp; &nbsp; }&nbsp; &nbsp; return rowRes{&nbsp; &nbsp; &nbsp; &nbsp; index:&nbsp; index,&nbsp; &nbsp; &nbsp; &nbsp; result: &result,&nbsp; &nbsp; }}func main() {&nbsp; &nbsp; numbers := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}&nbsp; &nbsp; matrix := [][]int{&nbsp; &nbsp; &nbsp; &nbsp; {1, 0, 0, 0, 0, 0, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 1, 0, 0, 0, 0, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 1, 0, 0, 0, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 1, 0, 0, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 1, 0, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 0, 1, 0, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 0, 0, 1, 0, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 0, 0, 0, 1, 0, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 0, 0, 0, 0, 1, 0},&nbsp; &nbsp; &nbsp; &nbsp; {0, 0, 0, 0, 0, 0, 0, 0, 0, 1},&nbsp; &nbsp; }&nbsp; &nbsp; rmatrix := make([][]int, 10)&nbsp; &nbsp; // Buffered channel&nbsp; &nbsp; rowChan := make(chan rowRes, 10)&nbsp; &nbsp; wg := sync.WaitGroup{}&nbsp; &nbsp; // Reciever goroutine&nbsp; &nbsp; go recv(rowChan, rmatrix)&nbsp; &nbsp; for i := range matrix {&nbsp; &nbsp; &nbsp; &nbsp; wg.Add(1)&nbsp; &nbsp; &nbsp; &nbsp; go func(index int, row []int, w *sync.WaitGroup) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; rowChan <- addRow(index, row, numbers)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; w.Done()&nbsp; &nbsp; &nbsp; &nbsp; }(i, matrix[i], &wg)&nbsp; &nbsp; }&nbsp; &nbsp; wg.Wait()&nbsp; &nbsp; close(rowChan)&nbsp; &nbsp; fmt.Println(rmatrix)}func recv(res chan rowRes, rmatrix [][]int) {&nbsp; &nbsp; for {&nbsp; &nbsp; &nbsp; &nbsp; select {&nbsp; &nbsp; &nbsp; &nbsp; case k, ok := <-res:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if !ok {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; rmatrix[k.index] = *k.result&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }}

Answer 3

流水线法taskChannel := make(chan string,1000); // Set up the task queuewg := sync.WaitGroup// Task releasewg.add(1)go func(&wg,taskChannel) {&nbsp; &nbsp; &nbsp; defer wg.Down()&nbsp; &nbsp; &nbsp; for i in "task list" {&nbsp; &nbsp; &nbsp; &nbsp; taskChannel <- "Stuff the characters you want to deal with here"&nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; // After the task is sent and closed&nbsp; &nbsp; close(taskChannel)}(wg *sync.WaitGroup,taskChannel chan string)// Task executiongo func(&wg,taskChannel,1000) {&nbsp; &nbsp; defer wg.Down()&nbsp; &nbsp; limit := make(chan bool,limitNumber); // Limit the number of concurrent&nbsp; &nbsp; tg := sync.WaitGroup&nbsp; &nbsp; loop:&nbsp; &nbsp; for {&nbsp; &nbsp; &nbsp; select {&nbsp; &nbsp; &nbsp; case task,over := <-taskChannel:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if !over {&nbsp; // If there are no more tasks, quit&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; tg.Wait()&nbsp; // Wait for all tasks to end&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; break loop&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; tg.Add(1)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; limit<-true&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; go func(&tg,limitm) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; defer func() {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; <-limit&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; tg.Down()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // Business processing logic, processing tasks&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }(tg *sync.WaitGroup,limit chan bool,task string)&nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }}(wg *sync.WaitGroup,taskChannel chan string,limitNumber int)wg.Wait()希望能帮到你