处理Go中生产者和消费者速率不匹配时如何同步关闭通道？

哈喽！今天心血来潮给大家带来了《处理Go中生产者和消费者速率不匹配时如何同步关闭通道？》，想必大家应该对Golang都不陌生吧，那么阅读本文就都不会很困难，以下内容主要涉及到，若是你正在学习Golang，千万别错过这篇文章~希望能帮助到你！

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

生产者(1个协程) -> chan0 -> 消费者生产者-1(>1个协程) -> chan1 -> 消费者-生产者-2(>1个协程) -> chan2 -> 消费者(>1个协程)

这里最多可以有 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".
   }

但即使采用这种方法，如果没有任何可用的内容，例如 10 秒，我也无法想出任何与关闭“input_messages”通道相同的方式。

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

1. 一旦第一个“chan0”关闭，所有后续通道都将关闭。
2. 所有生产者都会在关闭其输出通道之前进行更新，并且只有在所有生产者都完成写入后才会关闭通道。
3. 如果消费者在指定的超时时间内没有从通道获取数据，则应将其视为已关闭，并自行解除阻塞。

解决方案

使用 sync.waitgroup 来跟踪正在运行的 goroutine 的数量。每个 goroutine 在不再从通道获取数据后退出。一旦 waitgroup 完成，就可以完成清理工作。

类似这样的事情：

import (
        "sync"
        "time"
)

type Data interface{} // just an example

type Consumer interface {
        Consume(Data) Data
        CleanUp()
        Count() int
        Timeout() time.Duration
}

func StartConsumers(consumer Consumer, inCh <-chan Data, outCh chan<- Data) {
        wg := sync.WaitGroup{}
        for i := 0; i < consumer.Count(); i++ {
                wg.Add(1)
                go func() {
                consumeLoop:
                        for {
                                select {
                                case v, ok := <-inCh: // 'ok' says if the channel is still open
                                        if !ok {
                                                break consumeLoop
                                        }
                                        outCh <- consumer.Consume(v)
                                case <-time.After(consumer.Timeout()):
                                        break consumeLoop
                                }
                        }

                        wg.Done()
                }()
        }
        wg.Wait()

        consumer.CleanUp()
        close(outCh)
}

在管道的每个阶段，您都可以使用与上述类似的过程来启动消费者。

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《处理Go中生产者和消费者速率不匹配时如何同步关闭通道？》文章吧，也可关注golang学习网公众号了解相关技术文章。