repository·main·Indexed 21 days ago
https://github.com/wagslane/go-rabbitmqA high-level Go wrapper for the rabbitmq/amqp091-go library that simplifies RabbitMQ interactions. It provides automatic reconnection logic, multithreaded consumers, and sane defaults for queue and exchange declaration. The library includes support for clustered connections via NewClusterConn, functional configuration options for connection and consumer settings, and a structured Action system (Ack, NackDiscard, NackRequeue) for message handling.
When working with go-rabbitmq, follow these lifecycle rules to avoid resource leaks or errors:
Publisher and Consumer instances when they are no longer needed.rabbitmq.Conn once all associated publishers and consumers have been closed.Understanding the default behavior of go-rabbitmq helps in configuring your topology correctly:
rabbitmq.WithConsumerOptionsRoutingKey option.rabbitmq.WithConsumerOptionsExchangeDeclare (for consumers)rabbitmq.WithPublisherOptionsExchangeDeclare (for publishers)To consume messages, create a connection using rabbitmq.NewConn and then create a consumer using rabbitmq.NewConsumer.
Key behaviors:
rabbitmq.WithConsumerOptionsExchangeDeclare if you want the consumer to declare the exchange.consumer.Run to start processing messages. The handler function must return a rabbitmq.Action to indicate how the message should be handled.Available rabbitmq.Action values:
rabbitmq.Ack: Acknowledge the message.rabbitmq.NackDiscard: Negative acknowledgment, discard the message.rabbitmq.NackRequeue: Negative acknowledgment, requeue the message.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)
}To use go-rabbitmq in your Go project, install it using go get inside your Go module:
go get github.com/wagslane/go-rabbitmqTo publish messages, create a connection using rabbitmq.NewConn and then create a publisher using rabbitmq.NewPublisher.
Key behaviors:
rabbitmq.WithPublisherOptionsExchangeDeclare to ensure the exchange exists.publisher.Publish to send messages. You can specify the body, routing keys, and content type.Note: Always close your publishers and consumers before closing the connection.
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)
}When publishing with the mandatory or immediate flags set, the server may return messages that could not be routed. Use NotifyReturn to register a handler that receives these Return events.
Note: These notifications are shared across the entire connection. If you have multiple publishers on the same connection, they will all receive these returns.
publisher.NotifyReturn(func(r rabbitmq.Return) {
fmt.Printf("Message returned: %v\n", r.ReplyText)
})The go-rabbitmq library allows you to provide your own logging implementation by satisfying the Logger interface. You can inject your custom logger into the library using the following option functions:
WithPublisherOptionsLogger(logger Logger): Sets the logger for the Publisher.WithConsumerOptionsLogger(logger Logger):
Sets the logger for the Consumer.The Logger type is an alias for logger.Logger (from the internal package) and requires the implementation of the following methods:
Fatalf(format string, v ...interface{})Errorf(format string, v ...interface{})Warnf(format string, v ...interface{})Infof(format string, v ...interface{})Debugf(format string, v ...interface{})// Example of implementing the Logger interface
type MyLogger struct{}
func (m *MyLogger) Fatalf(format string, v ...interface{}) { /* implementation */ }
func (m *MyLogger) Errorf(format string, v ...interface{}) { /* implementation */ }
func (m *MyLogger) Warnf(format string, v ...interface{}) { /* implementation */ }
func (m *MyLogger) Infof(format string, v ...interface{}) { /* implementation */ }
func (m *MyLogger) Debugf(format string, v ...interface{}) { /* implementation */ }
// Usage with options
// publisher, err := rabbitmq.NewPublisher(amqpURL, rabbitmq.WithPublisherOptionsLogger(&MyLogger{}))You can control how the consumer logs information using the Logger field in ConsumerOptions.
WithConsumerOptionsLogging(options) to use a default logger that writes to stdout.WithConsumerOptionsLogger(log logger.Logger) to provide your own implementation of the logger.Logger interface.Exchanges and their bindings to queues are configured via the ExchangeOptions slice within ConsumerOptions.
direct, topic, fanout, headers), durability, and whether it is an internal exchange using functions like WithConsumerOptionsExchangeName, WithConsumerOptionsExchangeKind, and WithConsumerOptionsExchangeDurable.WithConsumerOptionsRoutingKey(routingKey) to bind the queue to a specific routing key on the primary exchange.WithConsumerOptionsBinding(binding) to add additional bindings with specific BindingOptions (like Declare).WithConsumerOptionsExchangeOptions(exchangeOptions) to add additional exchanges, allowing a single consumer to consume from multiple exchanges.To stop a consumer, use Close() or CloseWithContext(ctx).
Close(): By default, this waits for all active handlers to finish processing their current messages before returning (if CloseGracefully is enabled in options).CloseWithContext(ctx): Allows you to provide a context to limit how long the consumer waits for handlers to complete. If the context expires, the consumer will proceed with cleanup regardless of whether handlers have finished.Note that Close only stops the subscription and the consuming goroutines; it does not close the underlying connection manager.
// Graceful shutdown
consumer.Close()
// Shutdown with timeout
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
consumer.CloseWithContext(ctx)The *Conn object manages the underlying connection lifecycle.
Close(): Closes the connection and stops the connection manager. It is not safe to reuse a connection after calling Close(). You should close all consumers and publishers before closing the connection. It uses sync.Once to ensure multiple calls to Close() are safe.IsClosed(): Returns true if the connection is currently closed.if !conn.IsClosed() {
conn.Close()
}The go-rabbitmq library uses the functional options pattern to configure connections. You can pass one or more functions to your connection constructor (e.g., NewPublisher or NewConsumer) to customize the ConnectionOptions struct.
Key configuration options include:
WithConnectionOptionsBaseReconnectInterval(interval time.Duration): Sets the base reconnection interval. Consecutive failed attempts back off exponentially from this value, capped at 16x, with up to 25% jitter.WithConnectionOptionsLogger(log Logger): Sets a custom logger implementation.WithConnectionOptionsLogging(): A helper that sets the logger to the default stdDebugLogger which writes to stdout.WithConnectionOptionsConfig(cfg Config): Sets the Config used for the connection.// Example of using functional options to configure a connection
opts := []rabbitmq.Option{
rabbitmq.WithConnectionOptionsBaseReconnectInterval(time.Second * 2),
rabbitmq.WithConnectionOptionsLogging(),
rabbitmq.WithConnectionOptionsConfig(rabbitmq.Config{
// ... your config
}),
}
// These options would then be passed to a constructor like NewPublisher or NewConsumer
// publisher, err := rabbitmq.NewPublisher(opts...)