A wrapper of rabbitmq/amqp091-go that provides reconnection logic and sane defaults. Hit the project with a star if you find it useful ⭐
Supported by Boot.dev. If you'd like to learn about RabbitMQ and Go, you can check out my course here.
Streadway's AMQP library is currently the most robust and well-supported Go client I'm aware of. It's a fantastic option and I recommend starting there and seeing if it fulfills your needs. Their project has made an effort to stay within the scope of the AMQP protocol, as such, no reconnection logic and few ease-of-use abstractions are provided.
The goal with go-rabbitmq is to provide most (but not all) of the nitty-gritty functionality of Streadway's AMQP, but to make it easier to work with via a higher-level API. go-rabbitmq is also built specifically for Rabbit, not for the AMQP protocol. In particular, we want:
- Automatic reconnection
- Multithreaded consumers via a handler function
- Reasonable defaults
- Flow control handling
- TCP block handling
Inside a Go module:
go get github.com/wagslane/go-rabbitmqTake note of the optional options parameters after the queue name. The queue will be declared automatically, but the exchange will not. You'll also probably want to bind to at least one routing key.
conn, err := rabbitmq.NewConn(
"amqp://guest:guest@localhost",
rabbitmq.WithConnectionOptionsLogging,
)
if err != nil {
log.Fatal(err)
}
defer conn.Close()
consumer, err := rabbitmq.NewConsumer(
conn,
"my_queue",
rabbitmq.WithConsumerOptionsRoutingKey("my_routing_key"),
rabbitmq.WithConsumerOptionsExchangeName("events"),
rabbitmq.WithConsumerOptionsExchangeDeclare,
)
if err != nil {
log.Fatal(err)
}
defer consumer.Close()
err = consumer.Run(func(d rabbitmq.Delivery) rabbitmq.Action {
log.Printf("consumed: %v", string(d.Body))
// rabbitmq.Ack, rabbitmq.NackDiscard, rabbitmq.NackRequeue
return rabbitmq.Ack
})
if err != nil {
log.Fatal(err)
}The exchange is not declared by default, that's why I recommend using the following options.
conn, err := rabbitmq.NewConn(
"amqp://guest:guest@localhost",
rabbitmq.WithConnectionOptionsLogging,
)
if err != nil {
log.Fatal(err)
}
defer conn.Close()
publisher, err := rabbitmq.NewPublisher(
conn,
rabbitmq.WithPublisherOptionsLogging,
rabbitmq.WithPublisherOptionsExchangeName("events"),
rabbitmq.WithPublisherOptionsExchangeDeclare,
)
if err != nil {
log.Fatal(err)
}
defer publisher.Close()
err = publisher.Publish(
[]byte("hello, world"),
[]string{"my_routing_key"},
rabbitmq.WithPublishOptionsContentType("application/json"),
rabbitmq.WithPublishOptionsExchange("events"),
)
if err != nil {
log.Println(err)
}See the examples directory for more ideas.
- By default, queues are declared if they didn't already exist by new consumers
- By default, routing-key bindings are declared by consumers if you're using
WithConsumerOptionsRoutingKey - By default, exchanges are not declared by publishers or consumers if they don't already exist, hence
WithPublisherOptionsExchangeDeclareandWithConsumerOptionsExchangeDeclare.
Read up on all the options in the GoDoc, there are quite a few of them. I try to pick sane and simple defaults.
Close your publishers and consumers when you're done with them and do not attempt to reuse them. Only close the connection itself once you've closed all associated publishers and consumers.
Note that the API is currently in v0. I don't plan on huge changes, but there may be some small breaking changes before we hit v1.
Submit an issue here on GitHub
My goal is to keep dependencies limited to 1, github.com/rabbitmq/amqp091-go.
I would love your help! Contribute by forking the repo and opening pull requests. Please ensure that your code passes the existing tests and linting, and write tests to test your changes if applicable.
All pull requests should be submitted to the main branch.