streadway/amqp Connection/Channel does not reconnect if rabbitmq server restart/down and no cluster supported.
To simply developers, here is auto reconnect wrap with detail comments.
Also, with some basic helper functions like Connect
, DeclareQueues
, DeclareExchanges
This library depends on streadway/amqp
and sirius1024/go-amqp-reconnect
- add import
import "github.com/blackhades/go-amqp-reconnect/rabbitmq"
- Replace
amqp.Connection
/amqp.Channel
withrabbitmq.Connection
/rabbitmq.Channel
go run example/connect/demo.go
go run example/close/demo.go
go run example/reconnect/demo.go
import "github.com/blackhades/go-amqp-lib/rabbitmq"
rabbitmq.Connect("amqp://usr:[email protected]:5672/vhost,amqp://usr:[email protected]:5673/vhost,amqp://usr:[email protected]:5674/vhost"})
url := "amqp://127.0.0.1:5672,amqp://127.0.0.1:5672,amqp://127.0.0.1:5672,amqp://127.0.0.1:5672"
channel := rabbitmq.Connect(url)
if channel == nil {
log.Panic("Unable to connect to rabbitmq")
}
queues := []string{
"test-queue-1",
"test-queue-2",
"test-queue-3",
}
//Declare Multiple Queues with the same configuration
err := rabbitmq.DeclareQueues(queues, true, false, false, false, nil)
if err != nil {
log.Panic(err)
}
//this will push to the queue every 3 seconds
...
go func() {
for {
err = rabbitmq.PublishToQueue(queueName, "Pushing to a Queue Using Helper - " + time.Now().String())
log.Printf("publish, err: %v", err)
time.Sleep(3 * time.Second)
}
}()
url := "amqp://127.0.0.1:5672,amqp://127.0.0.1:5672,amqp://127.0.0.1:5672,amqp://127.0.0.1:5672"
channel := rabbitmq.Connect(url)
if channel == nil {
log.Panic("Unable to connect to rabbitmq")
}
exchanges := []string{
"test-exchange-1",
"test-exchange-2",
"test-exchange-3",
}
//Declare Multiple Exchanges with the same configuration
err := rabbitmq.DeclareExchanges(exchanges, amqp.ExchangeFanout, true, false, false, false, nil)
if err != nil {
log.Panic(err)
}
go func() {
for {
err = rabbitmq.PublishToExchange(exchangeName, "Pushing to an Exchange using helper - " + time.Now().String())
log.Printf("publish, err: %v", err)
time.Sleep(5 * time.Second)
}
}()
//Declare the function to process the data
func ProcessQueueData(byte []byte) error {}
//Map the queue to the function
go rabbitmq.Listen(map[string]rabbitmq.GenericFN{
"test-queue": ProcessQueueData,
}, 1)
- create the dead letter queue
- if an error is thrown when processing, the data goes to a deadletter queue.
rabbitmq.SetDeadLetterQueue("dead-letter-queue")
NOTE: when this function is called, data in the dead letter queue are sent back to the main queue they came from to be reprocessed.
rabbitmq.Listen(map[string]rabbitmq.GenericFN{
"dead-letter-queue": rabbitmq.ProcessDeadLetter,
}, prefetch)
You can set reconnect delay seconds with environment variable "AMQP_RECONNECTION_DELAY_IN_SECONDS" e.g.
# Default to 3s.
export AMQP_RECONNECTION_DELAY_IN_SECONDS=10