1 Receiver N Senders

• Fan-In
• 一個接收端, 多個發送端.
• 接收端發送關閉給一個額外的信號通道( stopCh := make(chan struct{}) ). 來通知發送端停止發送值.

Exercises

https://go.dev/play/p/0n63YY6E97k

在這個例子中,通道dataCh永遠不用關閉。是的,通道不必關閉. 如果沒有 goroutines 再引用它,一個通道最終將被垃圾回收,無論它是否關閉. 所以在這裡關閉通道的優雅不是關閉通道

package main

import (
    "log"
    "math/rand"
    "sync"
    "time"
)

func main() {
    rand.Seed(time.Now().UnixNano())
    log.SetFlags(0)

    // ...
    const Max = 100000
    const NumSenders = 1000

    wgReceivers := sync.WaitGroup{}
    wgReceivers.Add(1)

    // ...
    dataCh := make(chan int)
    stopCh := make(chan struct{})
    // stopCh is an additional signal channel.
    // Its sender is the receiver of channel
    // dataCh, and its receivers are the
    // senders of channel dataCh.

    // senders
    for i := 0; i < NumSenders; i++ {
        go func() {
            for {
                // The try-receive operation is to try
                // to exit the goroutine as early as
                // possible. For this specified example,
                // it is not essential.
                select {
                case <-stopCh:
                    return
                default:
                }

                // Even if stopCh is closed, the first
                // branch in the second select may be
                // still not selected for some loops if
                // the send to dataCh is also unblocked.
                // But this is acceptable for this
                // example, so the first select block
                // above can be omitted.
                select {
                case <-stopCh:
                    return
                case dataCh <- rand.Intn(Max):
                }
            }
        }()
    }

    // the receiver
    go func() {
        defer wgReceivers.Done()

        for value := range dataCh {
            if value == Max-1 {
                // The receiver of channel dataCh is
                // also the sender of stopCh. It is
                // safe to close the stop channel here.
                close(stopCh)
                return
            }

            log.Println(value)
        }
    }()

    // ...
    wgReceivers.Wait()
}