There's not really a one "right" way of doing things, and all of the others are wrong. However, there's a number of things that are considered good practice, and WRT to those, your code can do with a bit of TLC.
To that end, I'll just run through your code line by line, leaving comments on both style, and things I deem to be missing, hopefully ending up with something that is more "idiomatic" go.
package main
import (
"log"
"github.com/streadway/amqp"
)
Yup, I'm going to comment on your imports. The standard way to organise the imports in golang is to have multiple sections (separated by a blank line). The order is: first standard packages, second group have packages local to the project, then a third group with your external packages. In this case, the amqp package most certainly is an external one, so I'd write:
import (
"context"
"log"
"github.com/streadway/amqp"
)
As you may have notices, I've also added the context package. It's a good idea to look into this one, especially once you start playing around with multiple routines. More on that later
func main() {
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
panic(err)
}
defer conn.Close()
Ok, something you've probably heard before is "Don't panic". Maybe you know the phrase from the hitchhikers guide to the galaxy, but it's also one of the go proverbs. Just do a log.Fatalf("failed to connect to AMQP: %+v", err) or something. Fatal log is basically logging the output, and calling os.Exit(1).
for _, q := range []string{"q1", "q2", "q3"} {
go RunFetcher(q, conn, 3, 50)
}
select {}
}
The remainder of your main function just looks really weird. You seem to know what queues you want to consume from, but you've not given yourself any way to control the execution of the routines. Something you really would want is a way to stop the routines dead in their tracks. For example: the worker receives a kill signal, what you really want to do is cleanly terminate the workers. That's why I added the context package, because this allows you to do just that:
ctx, cfunc := context.WithCancel(context.Background())
defer cfunc() // cancel execution context when the main function returns
for _, q := range queueNames {
go runFetcher(ctx, q, 3, 50) // no need to export this func
}
Now let's add the bit that listens for a kill/interrupt signal:
import (
"os"
"os/signal"
)
func main() {
// add this:
sch := make(chan os.Signal, 1)
// listen for interrupt and kill signals
signal.Notify(sch, os.Interrupt, os.Kill)
// after you've done everything you needed to do:
<-sch // this is blocking, so after the read, the context will be automatically cancelled thanks to the defer cfunc() we added
}
Now your worker can run indefinitely, but cleans up after itself once it receives an interrupt or kill signal. Also: we don't need the empty select {}, which looks really nasty to my eye...
func RunFetcher(queueName string, conn *amqp.Connection, workers, qos int) {
ch, err := conn.Channel()
if err != nil {
log.Println(err.Error())
return
}
ch.Qos(qos, 0, false)
defer ch.Close()
msgs, err := ch.Consume(queueName, "", false, false, false, false, nil)
if err != nil {
log.Println(err.Error())
return
}
for index := 0; index < workers; index++ {
go func() {
for d := range msgs {
// process message
d.Ack(false)
}
}()
}
select {}
}
Ok, this is a big one to unpack. First off, let's not export this function (there's no reason to), and add the context argument as first argument to the function. Something I can't quite wrap my head around is why this function is kicked off in a routine, and doesn't even have a method of communicating any of the errors that it may encounter (for example in conn.Channel()). An error here causes an early return. To me, that error looks like something that needs to be handled. What's more, within this routine, you're actually spawning 3 subroutines that actually read from the same channel. That's fine, but it's actually this part you want to have running asynchronously. Why not change the runFetcher function to a normal function that returns an error, so you can handle it in main, and have it spawn as many routines as needed? You could also simply call conn.Channel() in the main function, and pass the Channel to the fetcher.
I'd also consider adding a loop so your fetchers can run indefinitely:
func runFetcher(ctx context.Context, queueName string, conn *amqp.Connection, workers, qos int) error {
ch, err := conn.Channel()
if err != nil {
return err
}
// the usual stuff, but:
msgs, err := ch.Consume(queueName, uuid, false, false, false, false, nil)
I would use a UUID package to create a unique consumer name, so we can actually stop our consumers correctly:
// create your workers like so:
for i := 0; i < workers; i++ {
go func() {
select {
case <-ctx.Done():
return // the context was cancelled, stop working
case msg := <- msgs:
msg.Ack(false) // acknowledge (or not)
}
}()
}
// now let's add this to stop the consumer
go func() {
<-ctx.Done()
ch.Cancel(uuid, false) // stop consumer quickly
}()
}
Now we're able to stop the consumers from actually getting messages from the queue, we can stop all our routines cleanly, and errors when establishing a consumer channel are actually propagated to the main function, so we can decide whether or not we want to proceed consuming messages from other queues or not.
These are some initial thoughts after quickly reading through the code. I might revisit this later on with more comments.
mainfunction but lackspackageandimportparts. You may edit your question to add them. \$\endgroup\$