asynchronous - 在golang中，如何编写一个为下一阶段引入延迟的流水线阶段？

Question

我正在关注 https://blog.golang.org/pipelines文章实现了几个阶段。

我需要其中一个阶段在事件传递到管道的下一阶段之前引入几秒钟的延迟。

我对下面的代码的担忧是，它会在传递事件之前产生无限数量的 time.Sleep() 例程。有没有更好的方法来做到这一点？

谢谢!

func fooStage(inChan <- chan *Bar) (<- chan *Bar) {
    out := make(chan *Bar, 10000)
    go func() {
        defer close(out)
        wg := sync.WaitGroup{}
        for {
            select {
            case event, ok := <-inChan:
                if !ok {
                    // inChan closed
                    break
                }
                wg.Add(1)
                go func() {
                    time.Sleep(5 * time.Second)
                    out <- event
                    wg.Done()
                }()
            }
        }
        wg.Wait()
    }()
    return out
}

Answer 1

您可以使用另一个 channel 来限制您的循环能够创建的事件 goroutine 的数量。

const numRoutines = 10

func fooStage(inChan <-chan *Bar) <-chan *Bar {
    out := make(chan *Bar, 10000)
    routines := make(chan struct{}, numRoutines)
    go func() {
        defer close(out)
        wg := sync.WaitGroup{}
        for {
            select {
            case event, ok := <-inChan:
                if !ok {
                    // inChan closed
                    break
                }
                wg.Add(1)
                routines <- struct{}{}
                go func() {
                    time.Sleep(5 * time.Second)
                    out <- event
                    wg.Done()
                    <-routines
                }()
            }
        }
        wg.Wait()
    }()
    return out
}