生产者速度慢，消费者速度快的情况下如何处理通道关闭同步？

Question

我是新手，找不到这个问题的答案。我正在做的是在生产者中读取 CSV 文件，做一些可能需要时间的事情，然后通过通道将输出发送给消费者。有一个生产者-消费者链，任何生产者最终都可能比它的消费者慢。

producer(1 goroutine) -> chan0 -> consumer-producer-1(>1 goroutines) -> chan1 -> consumer-producer-2(>1 goroutines) -> chan2 -> consumer(>1 goroutines)

这里最多可以有 15 个消费者。

现在我面临的问题是，如果生产者完成了，如何在消费者方面做出决定，我们可以停止处理。

我需要实现的是：

1. 一旦生产者完成，所有消费者最终应该做一些清理并在完成剩余的后退出

2. 如果消费者在特定的超时期限内没有获得任何数据，它可以退出（最好有一个信号）而不会进一步阻塞。

3. 它发生在序列中的所有生产者-消费者对上。

我使用了以下方法。

1. 将信号通道与每个数据通道保持在一起，并为其下一个消费者的每个 goroutine 发布“完成”。

2. 读取后，每个消费者应该只读取通道中剩余的缓冲数据，然后将 5“完成”放在下一个信号通道上。确保每个 goroutine 只有 5 个，而不是 5 个（使用https://golang.org/pkg/sync/#Once.Do）。

3. 以下是我到这里为止能想到的。

processRemaining = false

for processRemaining == false{

        select {

        case stuff, ok := <-input_messages:

                do_stuff(stuff)

                if ok == false { // if channel has been closed

                    processRemaining = true

                }

                if result != nil {

                        //send to channel output_messages

                }

        case sig := <-input_signals: // if signaled to stopped.

                fmt.Println("received signal", sig)

                processRemaining = true

        default:

                fmt.Println("no activity")

        }

}

if processRemaining {

        for stuff := range input_messages {

                do_stuff(stuff)

                if result != nil {

                        //send to channel output_messages

                }

        }

        // send "output_routine" number of "done" to a channel "output_signals".

}

但即使在这种方法中，我也无法想出任何方式来表现与关闭的“input_messages”通道相同的方式，如果没有可用的时间，比如 10 秒。

这种方法有没有我忽略的问题。解决这个问题的可能方法（或并发模式）是什么？确保：

1. 一旦第一个“chan0”关闭，所有后续通道都会关闭。

2. 所有的生产者在关闭他们的输出通道之前都会被更新，只有当他们都完成了他们的写入时，通道才会关闭。

3. 如果消费者在指定的超时时间内没有从通道中获取数据，它应该将其视为已关闭，并自行解除阻塞。

Answer 1

使用 async.WaitGroup来跟踪正在运行的 goroutines 的数量。每个 goroutine 在不再从通道获取数据后退出。一旦WaitGroup完成，清理就可以完成了。是这样的：import (&nbsp; &nbsp; &nbsp; &nbsp; "sync"&nbsp; &nbsp; &nbsp; &nbsp; "time")type Data interface{} // just an exampletype Consumer interface {&nbsp; &nbsp; &nbsp; &nbsp; Consume(Data) Data&nbsp; &nbsp; &nbsp; &nbsp; CleanUp()&nbsp; &nbsp; &nbsp; &nbsp; Count() int&nbsp; &nbsp; &nbsp; &nbsp; Timeout() time.Duration}func StartConsumers(consumer Consumer, inCh <-chan Data, outCh chan<- Data) {&nbsp; &nbsp; &nbsp; &nbsp; wg := sync.WaitGroup{}&nbsp; &nbsp; &nbsp; &nbsp; for i := 0; i < consumer.Count(); i++ {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; wg.Add(1)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; consumeLoop:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; for {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; select {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; case v, ok := <-inCh: // 'ok' says if the channel is still open&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if !ok {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; break consumeLoop&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; outCh <- consumer.Consume(v)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; case <-time.After(consumer.Timeout()):&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; break consumeLoop&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; wg.Done()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }()&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; wg.Wait()&nbsp; &nbsp; &nbsp; &nbsp; consumer.CleanUp()&nbsp; &nbsp; &nbsp; &nbsp; close(outCh)}在管道的每个阶段，您都可以使用与上述类似的过程来启动消费者。