1
\$\begingroup\$

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)
    }
}
\$\endgroup\$
1
  • \$\begingroup\$ Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers. \$\endgroup\$
    – Mast
    Commented Jun 1, 2022 at 7:16

1 Answer 1

2
\$\begingroup\$

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.

\$\endgroup\$
3
  • \$\begingroup\$ thank you for the review. I have adjusted the code according to your feedback. \$\endgroup\$
    – Srikanth
    Commented Jun 1, 2022 at 6:22
  • \$\begingroup\$ @Srikanth If your changes are significant enough, you can post a new version of your code as a new question. Feel free to add a link in the new question pointing towards your older question for extra context. \$\endgroup\$
    – Mast
    Commented Jun 1, 2022 at 7:17
  • 1
    \$\begingroup\$ @Rish, the "capturing loop variable" is good general advice, however, note that it doesn't apply here: when you start a goroutine with go, the arguments are evaluated first in the calling goroutine, and strconv.Itoa(i) produces a new string value each time. Even if you do pass i directly as go worker(i) it's okay, because i is copied into the worker() arg each loop. It's only when you capture it in a closure without explicitly passing it that this is a problem. \$\endgroup\$
    – Ben Hoyt
    Commented Jun 10, 2022 at 4:11

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.