从Sarama中的errors通道读取数据的正确方法是什么？

Question

当我产生一条消息时，我正在使用Go中编写的Sarama库从错误通道读取。整个代码如下所示，它包含在一个函数中：

producer.AsyncProducer.Input() <- &sarama.ProducerMessage{Topic: topic, Key: nil, Value: sarama.ByteEncoder(message)}
go func() {
    for err := range saramaProducer.Errors() {
        if producer.callbacks.OnError != nil {
            producer.callbacks.OnError(err)
        }
    }
}()

根据我对go例程的理解，我的go例程将在Errors()通道上不断迭代，直到它收到一个通道。一旦我的函数执行完毕，有没有办法让它停止监听错误？

Answer 1

您可以使用另一个通道和一个select来使循环返回。

var quit chan struct{}
go func() {
    for {
        select {
        case err:=<-saramaProducer.Errors():
            //handle errors
        case <-quit:
            return
        }
    }
}
defer func() { quit<-struct{}{} }()

最初的for ... range循环在获得通道之前不会迭代通道。相反，它会阻塞，直到出现错误，处理它，然后再次等待新的错误，直到通道关闭或main返回。

上面的代码有一个小问题，当quit和错误信道都准备好时，select随机选择一个，因此可能导致单个错误丢失。如果这是值得处理的，只需将另一个带有default的switch放入该错误，然后放入return。