go - 如何关闭具有多个发送者的 channel ？

Question

我有一个情况，我需要扇出发送到同一 channel 的接收器:

func MessagesFromSQS(ctx context.Context, sqsClient sqsiface.SQSAPI) chan *sqs.Message {
    messages := make(chan *sqs.Message)

    go func() {
        defer close(messages)
        wg := sync.WaitGroup{}

        for i := 0; i < parallelSQSReaders; i++ {
            wg.Add(1)
            go func() {
                defer wg.Done()

                for {
                    select {
                    case <-ctx.Done():
                        return

                    default:
                        // ...

                        for _, message := range result.Messages {
                            messages <- message
                        }
                    }
                }
            }()
        }

        wg.Wait()
    }()

    return messages
}

对我来说这是有道理的。然而，竞争检测器提示不同的 goroutine 以及发送和关闭 channel 。我意识到负责发送的 goroutine 应该与关闭的 goroutine 相同，但是正确的方法是什么？

编辑/已解决:感谢您的回复。事实证明我没有正确读取竞争检测器堆栈跟踪。我认为我更改的代码引入了该错误，而不是在 SQS 模拟中发现错误。一旦我正确同步了 ReceiveMessage() 就很好了。

Answer 1

当您知道不会再有任何写入时，即当所有工作程序完成时，请关闭 channel 。

所以:

wg.Wait()
close(messages)

附注我会通过将上下文取消的轮询与您的 channel 写入结合到选择中来重组您的轮询，例如

for _, message := range result.Messages {

    select {
        case messages <- message:
        case <-ctx.Done():
            return
    }

}