让我们快速浏览一下 Go 中的扇出模式。一般来说,扇出用于同时执行多个任务。
例如,假设您有一个数据管道,并且您想要处理各个项目。我们可以使用 go 例程和通道在收到项目时将其拆分,然后处理各个项目(例如放入 dB)。
这是一个易于实现的简单模式;但你需要管理通道以防止死锁。
// produce is simulating our single input as a channel func produce(id int) chan int { ch := make(chan int) go func() { for i := 0; i < 10; i++ { ch <- rand.Intn(20) } fmt.Printf("producer %d done\n", id) close(ch) // this is important!!! }() return ch } func worker(id int, jobs chan int, wg *sync.WaitGroup) { for value := range jobs { odd := "even" if (value & 1) == 1 { odd = "odd" } fmt.Printf("worker: %d, got %d is %s\n", id, value, odd) } wg.Done() } func main() { inputCh := produce(1) numWorkers := 3 jobs := make(chan int) // split input into individual jobs go func() { for value := range inputCh { jobs <- value } close(jobs) // this is important!!! }() // fan-out var wg sync.WaitGroup for i := 0; i < numWorkers; i++ { wg.Add(1) go worker(i, jobs, &wg) } wg.Wait() fmt.Println("done") }
这里的主要思想是有一个数据序列需要由固定数量的worker来操作。
对于输入,我们创建一个随机数序列并将它们放入通道中。我们将他们转移到另一个渠道,工人们将从中夺走他们的“工作”。
在此示例中,并非绝对有必要将输入移至作业通道。我们可以轻松地让工作人员从输入通道中拉出;这里只是为了清楚起见而这样做。
然后我们将固定数量的工作线程作为 goroutine 发送出去。每个工作人员将从作业通道中拉出,直到没有更多数据需要处理,此时它向 WaitGroup 发出信号,表明其已完成。
主线程使用 WaitGroup 来确保它在所有工作人员完成之前不会完成,即所有作业都已处理完毕。
需要提到的一个关键点是,此模式不会对处理输入序列的顺序做出任何保证。在很多情况下这可能没问题。例如,输入序列是包含自己的时间戳的数据记录,目标是将记录存储在 dB 中。在这种情况下,扇出是可以接受的。
最后一点,一旦序列中的所有数据都已发送,您将看到一些有关关闭通道的注释。这很关键。一旦没有更多数据,从通道中提取的范围运算符就会休眠。您可以通过注释掉一个将导致死锁情况的 close() 语句来验证这一点。 Goroutine 和 Channel 非常强大,但你必须明智地使用它们。
你会做什么不同的事情?我们如何改进这个例子?请在下面留下您的评论。
谢谢!
这篇文章以及本系列所有文章的代码可以在这里找到
以上是扇出模式的详细内容。更多信息请关注PHP中文网其他相关文章!