Using a High-Level RabbitMQ Client in Golang

rabbit

The post Using a High-Level RabbitMQ Client in Golang first appeared on Qvault.

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.

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:pass@localhost")
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)
}

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:pass@localhost")
if err != nil {
    log.Fatal(err)
}
err = publisher.Publish([]byte("hello, world"), []string{"routing_key"})
if err != nil {
    log.Fatal(err)
}

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[string]interface{}

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
    },
)

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,
)

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,
)

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.

Thanks For Reading!

If you’re interested in furthering your CS career, take our computer science courses

Follow and hit us up on Twitter @q_vault if you have any questions or comments, and if we’ve made a mistake be sure to let us know so we can get it corrected!

Subscribe to our newsletter for more programming articles



source https://qvault.io/2021/03/10/connecting-to-rabbitmq-in-golang-easy/

Comments

Post a Comment

Popular posts from this blog

Why is Exclusive Or (XOR) Important in Cryptography?

Image
If you are getting into cryptography, or just trying to understand the fundamentals, you may have noticed that the exclusive-or operation is used quite often, especially in ciphers. What is XOR ( ⊕ )? XOR, or “exclusive or” operates on binary data. It returns true if both of its inputs are opposites (one false and one true), otherwise, it returns false. An example in go code would be something like: func exclusiveOr(a bool, b bool) bool { return a != b } XOR Cipher – The Perfect Cipher The XOR operation can be used as a simple cipher for encrypting and decrypting messages with a single key. This is known as symmetric encryption. It is interesting to note that if: The key is the same size as the message The key is kept secret and generated truly randomly Then the cipher is impossible to crack. This is known as a one time pad . However, a simple XOR shouldn’t be used in production due to the key length needing to be too long to be practical. Cipher Example A...
Read more

Base64 vs Base58 Encoding

Image
By Lane Wagner –  @wagslane  on Twitter Base64 is one of the most popular encoding formats for representing data. Have some binary data? Base64 encode it for convenient readability and parsing. Base58 is just another encoding format (with 58 characters instead of 64, and has gained popularity largely due to Bitcoin and other cryptocurrencies. When it comes to data encoding, there is typically a trade-off made between: Human Readability: Do humans have a good idea of what is being represented at a glance? Efficient data compression: How many bytes are used to represent the same data? Let’s rank Base58, Base64, and ASCII encoding against each other using these metrics! First: A Note on Binary All data is stored in a raw binary format on computers. These encoding formats (Base64, Base58, and ASCII) are just different ways of reading and writing binary data. For example, in Base64 the binary code 000000 represents the letter A , but in ASCII the binary 00000000 repres...
Read more

(Very) Basic Intro to Hash Functions (SHA-256, MD-5, etc)

Image
This is going to be a basic introduction to hash functions. I will assume most of my audience is here to gain an understanding of why hash functions are used and the basic idea of why they work. My goal is to explain it in a general sense, I will be omitting proofs and implementation details and instead focusing on the high-level principles. Why Use A Hash Function? Hash functions are used all over the internet in order to securely store passwords, find duplicate records, quickly store and retrieve data, and more. For example,  Qvault  uses hashes to extend master passwords into private encryption keys. A longer list of uses here:  https://en.wikipedia.org/wiki/Hash_function#Uses I want to focus on several important features of hash functions, arguably the most important features. Hash functions scramble data deterministically No matter the input, the output of a hash function always has the same size The original data can not be retrieved from the scrambled d...
Read more