最初我写了一个创建n个通道,然后将1个go-routine附加到每个通道,并让go-routine在无限循环中运行。这样可以确保我一次只处理固定数量的下载。
我意识到这不是通道应该使用的方式,如果现在我理解得正确的话,实际上应该有一个通道,有n个go-routines在该通道上接收。每个go-routine都处于无限循环状态,等待消息,当它接收到消息时,它将处理数据,执行所有它应该执行的操作,完成后它将等待下一条消息。这使我能够确保我一次只处理n个文件。我认为这是正确的方法。我相信这就是"扇出",对吗?
我不需要将这些进程合并在一起。一旦下载完成,它将回调远程服务来处理余下的过程。应用程序没有其他需要执行的操作。
好的,看一下代码:
func main() {
queue, err := ConnectToQueue() // This works fine...
if err != nil {
log.Fatalf("Could not connect to queue: %s\n", err)
}
msgChannel := make(chan sqs.Message, 10)
for i := 0; i < MAX_CONCURRENT_ROUTINES; i++ {
go processMessage(msgChannel, queue)
}
for {
response, _ := queue.ReceiveMessage(MAX_SQS_MESSAGES)
for _, m := range response.Messages {
msgChannel <- m
}
}
}
func processMessage(ch <-chan sqs.Message, queue *sqs.Queue) {
for {
m := <-ch
// Do something with message m
// Delete message from queue when we're done
queue.DeleteMessage(&m)
}
}
我这样做是否正确?我有n个正在运行的go例程(其中
MAX_CONCURRENT_ROUTINES
=n),并且在循环中我们将不断向单个通道传递消息。这是正确的方法吗?我需要关闭任何东西还是可以无限期地让它运行?我注意到的一件事是SQS正在返回消息,但一旦我已经将10条消息传递给
processMessage()
(10是通道缓冲区的大小),就不会再处理任何其他消息了。谢谢大家!
return
/continue
问题。感谢您关于信号量的建议; 我会阅读相关主题。我不需要这些工作人员返回,他们只需处理包含回调URI的消息,然后调用它并等待下一条消息即可。感谢您对sync.WaitGroup
和2元组返回通道接收表达式的指导。再次感谢您的帮助。 - Engineer81