Go Routine: Multiple Consumers to RabbitMQ

Question

Here I'm trying to create the multiple consumers of rabbitmq queue using the goroutines. Also, I'm trying to achieve a graceful shutdown. Is the following approach create any memory leaks? or is it the correct way to handle them?

package main

import (
    "fmt"
    "log"
    "os"
    "os/signal"
    "runtime"
    "strconv"
    "syscall"
    "time"

    amqp "github.com/streadway/amqp"
)

var notifyClose = make(chan *amqp.Error)

func MaxParallelism() int {
    maxProcs := runtime.GOMAXPROCS(0)
    numCPU := runtime.NumCPU()
    if maxProcs < numCPU {
        return maxProcs
    }
    return numCPU
}

// newConsumer is a function create a rabbitMQ consumer
func newConsumer(connectionString string, fetchCount int) (connection *amqp.Connection, channel *amqp.Channel) {
    connection, _ = amqp.Dial(connectionString)
    channel, _ = connection.Channel()
    channel.Qos(fetchCount, 0, false)

    channel.NotifyClose(notifyClose)
    channel.NotifyPublish(notifyConfirm)
    return
}

func worker(channel *amqp.Channel, done chan bool, queueName, consumerName string) {
    msgs, _ := channel.Consume(
        queueName,    // queue
        consumerName, // consumer
        false,        // auto-ack
        false,        // exclusive
        false,        // no-local
        false,        // no-wait
        nil,          // args
    )

    for m := range msgs {
        body := string(m.Body)
        log.Printf("Processing data %+v\n", body)
        log.Printf("Processing data %+v done\n", body)
        time.Sleep(5 * time.Second)
        m.Ack(false)
        log.Printf("Data %+v acked\n", body)
    }
    done <- true
}

func main() {
    defer log.Println("Program stopped successful")
    url := "amqp://user:password@localhost:5672/"
    queue := "my_queue"
    name := "consumer"
    fetchSize := 10

    maxWorkers := MaxParallelism()
    fmt.Println("Maximum workers:", maxWorkers)
    log.Printf("Connecting to %s queue %s fetch-size %d\n", url, queue, fetchSize)
    connection, channel := newConsumer(url, fetchSize)
    log.Printf("Consumer %s is subscribing queue %s\n", name, queue)

    defer connection.Close()
    defer channel.Close()
    defer log.Println("Closing qeueu channel and connection")

    done := make(chan bool, maxWorkers)
    for i := 0; i < maxWorkers; i++ {
        go worker(channel, done, queue, "consumer-"+strconv.Itoa(i))
    }

    exit := make(chan os.Signal, 1)
    signal.Notify(exit, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)

    select {
    case <-notifyClose:
        closeWorkes(channel)
    case <-exit:
        closeWorkes(channel)
        log.Println("Got exit signal")
        // Stop recieving message from queue
        log.Println("Stopped receiving message from queue")
        // Wait for worker procrss recieved message
        log.Println("Wait for worker procrss recieved message")
    }
    <-done
    log.Println("Woker done")
    // Just to print if all goroutines are closed or not
    // it is returning 4 and that is correct
    // the main goroutine
    // the background sweeper (the phase of the garbage collection which is concurrent)
    // the scavenger (also part of the garbage collector)
    // the finalizer goroutine (exclusively running the finalizers eventually attached to objects)
    time.Sleep(2 * time.Second)
    fmt.Println("Remaining goroutines", runtime.NumGoroutine())
}

func closeWorkes(channel *amqp.Channel) {
    maxWorkers := MaxParallelism()
    for i := 0; i < maxWorkers; i++ {
        channel.Cancel("consumer-"+strconv.Itoa(i), false)
    }
}

Answer 1

Is the following approach creating any memory leaks?

Go is a garbage-collected language, which means any allocations are cleaned up by the GC. Unless you're using Cgo and malloc-ing memory, you don't really have to worry too much about memory leaks.

---

Since MaxParallelism is used within the package, you can rename it maxParallelism.

---

connection, _ = amqp.Dial(connectionString)

You're not checking the error here. In fact, you're not checking errors anywhere. Ignoring errors is not recommended. In fact, using connection or channel without checking the errors may mean they can be nil, which will cause the program to panic when used.

---

I see in main that the caller of newConsumer has to call defer connection.Close() and defer channel.Close(). This is fine, I guess, for smaller programs. As programs get more complicated, errors are bound to creep in. What if the caller forgets to close a channel or connection or both? Suggest passing a function to the caller; the caller now has to explicitly ignore the returned value or defer the call to it. For example,

func newConsumer(...) (*amqp.Connection, *amqp.Channel, func()) {
    ...
    return connection, channel, func() {
            channel.Close()
            connection.Close()
    }
}

func main() {
     ...
     conn, channel, closeFn := newConsumer(...)
     defer closeFn() 
}

---

for i := 0; i < maxWorkers; i++ {
        go worker(channel, done, queue, "consumer-"+strconv.Itoa(i))
}

Beware of this trap. This might not behave as expected. The goroutine captures the value of the loop variable. By the time the goroutine gets scheduled, the value of i may not be the same.

A simple example can be found here: https://go.dev/play/p/HkDLewt-op_U

An easy way to fix this is to capture the variable inside the loop.

for i := 0; i < maxWorkers; i++ {
        i := i      // <------ this
        go worker(channel, done, queue, "consumer-"+strconv.Itoa(i))
}

Example: https://go.dev/play/p/tWU3KHcYOMk

---

Prefer using fmt.Sprintf("consumer-%d", i) over "consumer-"+strconv.Itoa(i)

---

In the examples I linked above, you might have seen me using sync.WaitGroup. This is the preferred way to 'wait' for goroutines to finish instead of manually creating a channel and sending messages.

If your goroutines may return an error (as they should because of the call to channel.Consume, prefer using errgroup.

---

Speaking of,

<- done

will only wait for a single message i.e. will only block until the first goroutine finishes. All the other goroutines may still be doing work.

---

I'm not sure what this line is trying to achieve:

// it is returning 4 and that is correct
...
fmt.Println("Remaining goroutines", runtime.NumGoroutine())

You should not try to make guesses about how many goroutines are currently active.

---

nit: closeWorkers, not closeWorkes.