Go 中的扇出-扇宁

PHPz

发布时间：2024-07-29 12:56:50

635人浏览过

来源于dev.to

转载

go 中的扇出-扇宁

在之前的 2 篇文章中，我们分别介绍了 fanout 和 fanin。通常情况下，我们将它们一起使用，因为我们有一个数据流，我们希望单独对项目进行操作，并且可以使用并发安全地执行此操作。因此，我们扇出到多个工作线程，然后扇回到单个流。

例如，假设您有一个很大的日志文件。您可以将文件分成块，允许每个工作人员同时操作文件的不同部分，然后合并结果。

如果您遵循前两篇文章，这种模式是显而易见的。如果您不确定，请参阅上面的链接。

// produce is simulating our single input as a channel
func produce() chan int {
    ch := make(chan int)
    go func() {
        for i := 0; i < 10; i++ {
            ch <- rand.Intn(50)
        }
        fmt.Printf("producer done\n")
        close(ch) // this is important!!!
    }()
    return ch
}

func worker(id int, jobs chan int, out chan OddEven, wg *sync.WaitGroup) {
    for value := range jobs {
        odd := "even"
        if (value & 1) == 1 {
            odd = "odd"
        }
        out <- OddEven{
            Number:  value,
            OddEven: odd,
        }
    }
    close(out) // remember this
    wg.Done()
}

// OddEven struct will be the result of the work done by each fanout thread
// and be the fanin data
type OddEven struct {
    Number  int
    OddEven string
}

func fanin(inputs []chan OddEven) chan OddEven {
    output := make(chan OddEven)

    var wg sync.WaitGroup
    for i, input := range inputs {
        wg.Add(1)
                // explicit params to capture loop vars
        go func(id int, input chan OddEven, output chan OddEven, wg *sync.WaitGroup) {
            for value := range input {
                output <- value
            }
            fmt.Printf("done merging source %d\n", id)
            wg.Done()
        }(i, input, output, &wg)
    }
    go func() {
        wg.Wait()
        close(output) // this is important!!!
    }()

    return output
}

func main() {
    // simulate the input data stream
    inputCh := produce()

    numWorkers := 3
    // fan-out to send data items to workers as individual jobs
    var wg sync.WaitGroup
    workerResults := make([]chan OddEven, numWorkers)
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        workerResults[i] = make(chan OddEven)
        go worker(i, inputCh, workerResults[i], &wg)
    }
    go func() {
        wg.Wait()
    }()

    // fan-in the results
    results := fanin(workerResults)

    done := make(chan bool)
    go func() {
        for value := range results {
            fmt.Printf("got %d is %s\n", value.Number, value.OddEven)
        }
        close(done)
    }()
    <-done
    fmt.Println("done")
}

有一个 produce() 函数可以创建模拟的数字输入流。

有一个工作函数在输入通道上运行，直到没有更多数据为止。它对每个值“处理”输入数据（确定该值是奇数还是偶数），然后将结果结构发送到输出通道。

请注意，当每个工作人员完成后，它会关闭其结果通道。这对于防止死锁是必要的，因为否则 fanin 操作将休眠等待 chan 上的更多数据。

主线程从 produce 获取输入流，然后启动许多工作线程，为每个工作线程提供自己的通道，以便发送结果。

这些结果通道随后被发送到 fanin 操作。为了扇入，我们创建一个通道来接收输出，然后为每个工作通道启动一个 goroutine。每个 goroutine 只是在通道上进行迭代，直到没有更多数据为止，然后终止。 请记住，我们关闭了工作线程中的结果通道，这就是允许 for 循环终止的原因