RabbitMQ Golang Cheat-Sheet

March 15, 2025•5 min read

RabbitMQ Docker setup

Run RabbitMQ with the management plugin enabled:

docker run -d \
  --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  rabbitmq:management

The connection string is amqp://localhost:5672

You can connect to your instance using RabbitGUI at http://localhost:15672 (guest/guest)

Docker compose setup for RabbitMQ

Create a docker-compose.yml file with custom credentials and persistent storage:

version: '3'
services:
  rabbitmq:
    image: rabbitmq:management
    container_name: rabbitmq
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: user
      RABBITMQ_DEFAULT_PASS: password
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq
 
volumes:
  rabbitmq_data:

The connection string is amqp://user:password@localhost:5672

You can connect to your instance using RabbitGUI at http://localhost:15672 (user/password)

Installation in Go

Add the amqp091-go package to your project:

go get github.com/rabbitmq/amqp091-go

Producer example

You can publish directly to a queue using the PublishWithContext method:

package main
 
import (
	"context"
	"log"
 
	amqp "github.com/rabbitmq/amqp091-go"
)
 
func main() {
	conn, err := amqp.Dial("amqp://localhost:5672")
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
 
	ch, err := conn.Channel()
	if err != nil {
		log.Fatal(err)
	}
	defer ch.Close()
 
	q, err := ch.QueueDeclare("my_queue", false, false, false, false, nil)
	if err != nil {
		log.Fatal(err)
	}
 
	err = ch.PublishWithContext(context.Background(),
		"",     // exchange
		q.Name, // routing key
		false,  // mandatory
		false,  // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte("Hello RabbitGUI!"),
		},
	)
	if err != nil {
		log.Fatal(err)
	}
}

Consumer example

Your consumer can listen to messages from a queue using the Consume method:

package main
 
import (
	"log"
 
	amqp "github.com/rabbitmq/amqp091-go"
)
 
func main() {
	conn, err := amqp.Dial("amqp://localhost:5672")
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
 
	ch, err := conn.Channel()
	if err != nil {
		log.Fatal(err)
	}
	defer ch.Close()
 
	q, err := ch.QueueDeclare("my_queue", false, false, false, false, nil)
	if err != nil {
		log.Fatal(err)
	}
 
	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
	if err != nil {
		log.Fatal(err)
	}
 
	for msg := range msgs {
		log.Printf("Message: %s", msg.Body)
	}
}

Exchanges examples

Direct

A direct exchange routes messages to queues based on exact routing key matches:

err = ch.ExchangeDeclare("my_exchange", "direct", false, false, false, false, nil)
if err != nil {
	log.Fatal(err)
}
 
q, err := ch.QueueDeclare("my_queue", false, false, false, false, nil)
if err != nil {
	log.Fatal(err)
}
 
err = ch.QueueBind(q.Name, "my.routing.key", "my_exchange", false, nil)
if err != nil {
	log.Fatal(err)
}
 
err = ch.PublishWithContext(context.Background(),
	"my_exchange",   // exchange
	"my.routing.key", // routing key
	false, false,
	amqp.Publishing{
		ContentType: "text/plain",
		Body:        []byte("Hello RabbitGUI!"),
	},
)
 
msgs, _ := ch.Consume(q.Name, "", true, false, false, false, nil)
for msg := range msgs {
	log.Printf("Message: %s", msg.Body)
}

Fanout

A fanout exchange broadcasts messages to all bound queues, ignoring routing keys:

err = ch.ExchangeDeclare("my_exchange", "fanout", false, false, false, false, nil)
if err != nil {
	log.Fatal(err)
}
 
q, err := ch.QueueDeclare("my_queue", false, false, false, false, nil)
if err != nil {
	log.Fatal(err)
}
 
err = ch.QueueBind(q.Name, "", "my_exchange", false, nil)
if err != nil {
	log.Fatal(err)
}
 
err = ch.PublishWithContext(context.Background(),
	"my_exchange", // exchange
	"",            // routing key
	false, false,
	amqp.Publishing{
		ContentType: "text/plain",
		Body:        []byte("Hello RabbitGUI!"),
	},
)
 
msgs, _ := ch.Consume(q.Name, "", true, false, false, false, nil)
for msg := range msgs {
	log.Printf("Message: %s", msg.Body)
}

Topic

A topic exchange routes messages based on wildcard pattern matching (* for one word, # for zero or more words):

err = ch.ExchangeDeclare("my_exchange", "topic", false, false, false, false, nil)
if err != nil {
	log.Fatal(err)
}
 
q, err := ch.QueueDeclare("my_queue", false, false, false, false, nil)
if err != nil {
	log.Fatal(err)
}
 
err = ch.QueueBind(q.Name, "*.routing.*", "my_exchange", false, nil)
if err != nil {
	log.Fatal(err)
}
 
err = ch.PublishWithContext(context.Background(),
	"my_exchange",    // exchange
	"my.routing.key", // routing key
	false, false,
	amqp.Publishing{
		ContentType: "text/plain",
		Body:        []byte("Hello RabbitGUI!"),
	},
)
 
msgs, _ := ch.Consume(q.Name, "", true, false, false, false, nil)
for msg := range msgs {
	log.Printf("Message: %s", msg.Body)
}

Message acknowledgements

Automatic acknowledgement

RabbitMQ will dequeue messages as soon as they've been sent down the wire.

msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
//                                    ^^^^
//                                    autoAck = true

Manual acknowledgement

Manually acknowledge messages after processing to ensure they are removed from the queue:

msgs, err := ch.Consume(q.Name, "", false, false, false, false, nil)
 
for msg := range msgs {
	// Process message...
	msg.Ack(false)
}

Reject and requeue

Reject a message and put it back in the queue for reprocessing:

msgs, err := ch.Consume(q.Name, "", false, false, false, false, nil)
 
for msg := range msgs {
	// Process message...
	msg.Nack(false, true) // multiple=false, requeue=true
}

Reject and do not requeue

Reject a message and discard it (sends to dead letter queue if configured):

msgs, err := ch.Consume(q.Name, "", false, false, false, false, nil)
 
for msg := range msgs {
	// Process message...
	msg.Nack(false, false) // multiple=false, requeue=false
}

Dead Letter Queues

Configure a queue with a dead letter exchange to catch rejected or expired messages:

package main
 
import (
	"log"
 
	amqp "github.com/rabbitmq/amqp091-go"
)
 
func main() {
	conn, err := amqp.Dial("amqp://localhost:5672")
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
 
	ch, err := conn.Channel()
	if err != nil {
		log.Fatal(err)
	}
	defer ch.Close()
 
	// Create the dead letter exchange and queue
	err = ch.ExchangeDeclare("dlx_exchange", "direct", false, false, false, false, nil)
	if err != nil {
		log.Fatal(err)
	}
 
	_, err = ch.QueueDeclare("dead_letter_queue", false, false, false, false, nil)
	if err != nil {
		log.Fatal(err)
	}
 
	err = ch.QueueBind("dead_letter_queue", "", "dlx_exchange", false, nil)
	if err != nil {
		log.Fatal(err)
	}
 
	// Create main queue with dead letter configuration
	args := amqp.Table{
		"x-dead-letter-exchange":    "dlx_exchange",
		"x-dead-letter-routing-key": "",
	}
 
	_, err = ch.QueueDeclare("my_queue", false, false, false, false, args)
	if err != nil {
		log.Fatal(err)
	}
}