Using a High-Level RabbitMQ Client in Golang

If you’ve already read my previous post, you know that the amqp package is awesome and you can get up and running with just 40-50 lines of simple code. Unfortunately, the bare-bones amqp library doesn’t handle a lot of the stuff you probably wish it did, things like reconnecting logic, the spawning of threads, queue and binding boilerplate, and flow control.

Fear not! I recently open-sourced my own package that neatly wraps Streadway’s amqp library and provides those higher-level abstractions. Behold, go-rabbitmq.

My main goals when building this new package were the following:

  • Automatic reconnection handling. The amqp lib doesn’t handle when a Node goes down.
  • The spawning of concurrent goroutines to consume from a queue – there’s no reason for everyone to have to write those same 20 lines of code in each app they build.
  • Reasonable defaults, but total control. Using go-rabbitmq, you still get practically all of the control you had at the amqp level, but defaults are provided that will reduce the amount of boilerplate code you need in your application logic.
  • Flow control. The AMQP library just provides a channel that tells you when publishers should stop. The go-rabbitmq library actually returns an error when publishing to a server that has requested the client to stop publishing.

Sorry to interrupt! I just wanted to mention that you should check out my new free Go course. It’s designed to teach you all the fundamentals of my favorite coding language.

Examples

The go-rabbitmq library provides two types that will hold all your configurations for publishing and consuming messages, a Publisher and a Consumer.

Consuming with the defaults

consumer, err := rabbitmq.NewConsumer("amqp://user:[email protected]") if err != nil { log.Fatal(err) } err = consumer.StartConsuming( func(d rabbitmq.Delivery) bool { log.Printf("consumed: %v", string(d.Body)) // true to ACK, false to NACK return true }, "my_queue", []string{"routing_key1", "routing_key2"} ) if err != nil { log.Fatal(err) }
Code language: Go (go)

The code above does the following:

  1. Creates a new Consumer connected to the provided cluster
  2. Creates the my_queue queue if it doesn’t exist and binds routing_key1 and routing_key2 to it.
  3. Starts consuming messages on a single thread using the provided handler. We provided a handler that prints the message and ACKs it off the queue.
  4. If the app loses connection to the cluster or a node goes down, the client will continuously try to reconnect with an exponential backoff strategy.

Publishing with the defaults

publisher, returns, err := rabbitmq.NewPublisher("amqp://user:[email protected]") if err != nil { log.Fatal(err) } err = publisher.Publish([]byte("hello, world"), []string{"routing_key"}) if err != nil { log.Fatal(err) }
Code language: Go (go)

The code above does the following:

  1. Creates a publisher connected to the provided cluster
  2. Creates a returns channel that will send messages back when the server rejects a publish. Note that this happens when specific options (mandatory and immediate) are used. If the cluster goes down or a connection is lost, publishing will fail until the publisher reconnects, which it does automatically.
  3. Publish the bytes of the text “hello, world” to the routing_key routing key.

Consuming Options

The full suite of consuming options is quite large, I didn’t want to limit functionality:

type ConsumeOptions struct { QueueDurable bool QueueAutoDelete bool QueueExclusive bool QueueNoWait bool QueueArgs Table BindingExchange string BindingNoWait bool BindingArgs Table Concurrency int QOSPrefetch int QOSGlobal bool ConsumerName string ConsumerAutoAck bool ConsumerExclusive bool ConsumerNoWait bool ConsumerNoLocal bool ConsumerArgs Table } type Table map[<a href="https://pkg.go.dev/builtin#string">string</a>]interface{}
Code language: Go (go)

Most of the options are fairly self-explanatory if you look into the feature set of RabbitMQ. For example, durable queues aren’t lost on server restart, exclusive queues can’t be used by more than one connection, auto-delete queues are deleted when they have no consumers, etc. Let’s go over how to set some of the more advanced configurations.

err = consumer.StartConsuming( func(d rabbitmq.Delivery) bool { log.Printf("consumed: %v", string(d.Body)) return true }, "my_queue", []string{"routing_key1", "routing_key2"}, func(opts *rabbitmq.ConsumeOptions) { opts.QueueDurable = true opts.Concurrency = 10 opts.QOSPrefetch = 100 }, )
Code language: Go (go)

All that changes is that we pass in function(s) (using Go’s variadic feature) that will mutate a pointer to the configuration struct the way we want. In the above code, we made our queue durable, we tell the consumer we want ten threads running their own handler, and we want the server to send us batches of 100 messages at a time (this just helps with throughput, each handler still only processes a single message at once).

There are also configuration functions provided that you can use directly:

err = consumer.StartConsuming( func(d rabbitmq.Delivery) bool { log.Printf("consumed: %v", string(d.Body)) return true }, "my_queue", []string{"routing_key1", "routing_key2"}, rabbitmq.WithConsumeOptionsConcurrency(10), rabbitmq.WithConsumeOptionsQueueDurable, rabbitmq.WithConsumeOptionsQuorum, )
Code language: Go (go)

Publishing options

Frankly, I’m not a huge fan of the mandatory and immediate options. To me, it seems like it breaks a simple tenet of the pub/sub architecture, that is, publishers should be unaware of their consumers. It should be up to the consumers to make sure they’re correctly connected and their queues are created. That said, you can still use those options:

err = publisher.Publish( []byte("hello, world"), []string{"routing_key"}, // leave blank for defaults rabbitmq.WithPublishOptionsContentType("application/json"), rabbitmq.WithPublishOptionsMandatory, rabbitmq.WithPublishOptionsPersistentDelivery, )
Code language: Go (go)

If you have any questions about the library, or suggestions for improvement please open and issue and let’s talk about it! I plan to keep the library in v0 until I’m super sure I’m happy with the API.

Have questions or feedback?

Follow and hit me up on Twitter @q_vault if you have any questions or comments. If I’ve made a mistake in the article be sure to let me know so I can get it corrected!

Leave a Comment