multithreading - 如何等待一组 goroutines 中的*任何*发出信号而不需要我们等待它们这样做

Question

有很多示例说明如何使用 WaitGroup 等待一组 goroutine 中的所有完成，但是如果你想等待其中的任何一个怎么办？他们在不使用信号量系统的情况下完成某些进程必须等待？例如，在生产者/消费者场景中，多个生产者线程向数据结构添加多个条目，而消费者一次删除一个条目。在这种情况下:

• 我们不能只使用标准的生产者/消费者信号量系统，因为生产:消费不是 1:1，也因为数据结构充当缓存，所以生产者可以“自由运行”而不是阻塞，直到消费者“准备好”消费他们的产品。
• 数据结构可能被消费者清空，在这种情况下，消费者希望等到任何一个生产者完成(这意味着数据结构中可能有新的东西)

问题:有没有标准的方法来做到这一点？

我只能想出两种方法来做到这一点。两者都通过使用 channel 作为信号量:

var unitary_channel chan int = make(chan int, 1)

func my_goroutine() {
   // Produce, produce, produce!!!
   unitary_channel<-0 // Try to push a value to the channel
   <-unitary_channel // Remove it, in case nobody was waiting
}

func main() {
   go my_goroutine()
   go my_goroutine()
   go my_goroutine()
   for len(stuff_to_consume) { /* Consume, consume, consume */ }
   // Ran out of stuff to consume
   <-unitary_channel
   unitary_channel<-0 // To unblock the goroutine which was exiting
   // Consume more
}

现在，这个简单的示例有一些明显的(但可以解决的问题)，例如如果没有至少一个 go_routine() 仍在运行，则 main() 无法退出这一事实。

第二种方法不是要求生产者删除他们刚刚推送到 channel 的值，而是使用select 允许生产者在 channel 阻塞时退出。

var empty_channel chan int = make(chan int)

func my_goroutine() {
   // Produce, produce, produce!!!
   select {
      case empty_channel <- 0: // Push if you can
      default:  // Or don't if you can't
   }
}

func main() {
   go my_goroutine()
   go my_goroutine()
   go my_goroutine()
   for len(stuff_to_consume) { /* Consume, consume, consume */ }
   // Ran out of stuff to consume
   <-unitary_channel
   // Consume more
}

当然，如果所有的 goroutines 都已经终止，这个函数也会永远阻塞 main() 。 那么，如果第一个问题的答案是“不，除了您提出的解决方案之外，没有其他标准解决方案”，是否有令人信服的理由说明为什么应该使用其中一个而不是另一个?

Answer 1

你可以像这样使用带有缓冲区的 channel

// create a channel with a buffer of 1
var Items = make(chan int, 1)
var MyArray []int

func main() {
    go addItems()
    go addItems()
    go addItems()
    go sendToChannel()
    for true {
        fmt.Println(<- Items)
    }
}

// push a number to the array
func addItems() {
    for x := 0; x < 10; x++ {
        MyArray = append(MyArray, x)
    }
}

// push to Items and pop the array
func sendToChannel() {
    for true {
        for len(MyArray) > 0 {
            Items <- MyArray[0]
            MyArray = MyArray[1:]
        }
        time.Sleep(10 * time.Second)
    }
}

main 中的 for 循环将永远循环并打印任何添加到 channel 的内容，并且 sendToChannel 函数将在数组为空时阻塞，

这样生产者就永远不会被阻塞，消费者可以在有一个或多个可用项目时消费