Handling RabbitMQ Reconnections With Go and amqp091-go
Published: 2024-06-11 09:15:42

In a distributed system, maintaining a stable connection between components is crucial for ensuring reliable communication. This article explores strategies for reconnecting to RabbitMQ, ensuring that your applications can recover from network disruptions or broker restarts without losing valuable data.

# Setting  Up
## Importing Dependencies
```go
import (
	"fmt"
	amqp "github.com/rabbitmq/amqp091-go"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"
)
```
We use `amqp091-go` as a connection for RabbitMQ
## Preparing option
```go
type Config struct {
	username string
	password string
	host     string
	port     int
}

type Container struct {
	connection *amqp.Connection
}

type Channel struct {
	name       string
	kind       string
	durable    bool
	autoDelete bool
	internal   bool
	noWait     bool
	args       amqp.Table
}
```
- `Config` holds connection configuration.
- `Container` holds the connection to RabbitMQ. In each application, they should have a wrapper for holding the connection.
- `Channel` holds channel configuration.

## Connection Creation
```go
// dial 
// create connection and register channel
func dial(config Config, channels []Channel) (conn *amqp.Connection, err error) {

	// create connection
	conn, err = amqp.Dial(fmt.Sprintf("amqp://%s:%s@%s:%d", config.username, config.password, config.host, config.port))
	if err != nil {
		return
	}
	log.Print("rabbitmq connect success")

	// register channel
	channel, err := conn.Channel()
	for _, c := range channels {
		err = channel.ExchangeDeclare(c.name, c.kind, c.durable, c.autoDelete, c.internal, c.noWait, c.args)
		if err != nil {
			return
		}
	}
	log.Print("rabbitmq channels register success")

	return
}
```
We create a Go function `dial` to assist in establishing a connection to RabbitMQ. It takes in a `Config` struct containing connection details like username, password, host, and port, and a slice of `Channel` structs representing different channels to be registered with the server.


## Connection Status
```go
func check(container *Container) {
	for {
		time.Sleep(time.Duration(1) * time.Second)
		log.Printf("rabbitmq connected: %v", !container.connection.IsClosed())
	}
}
```
We create a Go function `check` to check and print connection status periodically, to have visibility of the connection status.


# Reconnection Code
Below is the full code containing reconnection code. We will walk through step by step to understand how reconnection works in Go with RabbitMQ.
```go
func reconnect(container *Container, config Config, channels []Channel) {

	go func() {
		for {
			reason, ok := <-container.connection.NotifyClose(make(chan *amqp.Error))
			if !ok {
				log.Print("rabbitmq connection closed")
				break
			}
			log.Printf("rabbitmq connection closed unexpectedly, reason: %v", reason)

			for {

				time.Sleep(time.Duration(1) * time.Second)

				connection, err := dial(config, channels)

				if err == nil {
					container.connection = connection
					log.Print("rabbitmq reconnect success")
					break
				}

				log.Printf("rabbitmq reconnect failed, err: %v", err)
			}

		}
	}()
}
```
This function is a reconnect loop for a RabbitMQ client, and since reconnection creates a brand-new connection, the `reconnect` function is required to have all the values like the initial connection.

1. It starts a goroutine to prevent blocking that listens for the closure of the connection.
2. When the connection is closed, it enters a loop to attempt reconnection.
3. It uses a delay of 1 second (`time.Sleep(time.Duration(1) * time.Second)`) between reconnection attempts.
4. It attempts to reconnect by calling the `dial` function with the provided configuration and channels.
5. If the reconnection is successful, it updates the `container.connection` with the new connection.
6. If the reconnection fails, it logs the error and continues the loop.

## Stitching Together
```go
func main() {

	config := Config{
		username: "guest",
		password: "guest",
		host:     "localhost",
		port:     5672,
	}

	channels := []Channel{
		{
			name:       "reconnection-exchange",
			kind:       "direct",
			durable:    true,
			autoDelete: false,
			internal:   false,
			noWait:     false,
			args:       nil,
		},
	}

	container := &Container{}

	connection, err := dial(config, channels)
	if err != nil {
		log.Fatal(err)
		return
	}
	container.connection = connection
	go check(container)

	defer func() {
		if err := container.connection.Close(); err != nil {
			log.Print(err)
		}
		log.Print("rabbitmq connection closed")

	}()

	reconnect(container, config, channels)

	{
		quit := make(chan os.Signal, 1)
		signal.Notify(quit, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
		<-quit

	}
}
```
Now we combine all code together in `main` function.


# Testing
## Running
Before running the code, make sure RabbitMQ is running and accessible. Upon running the code, you should see a message like the one below.
```
rabbitmq connect success
rabbitmq channels register success
rabbitmq connected: true
rabbitmq connected: true
rabbitmq connected: true
rabbitmq connected: true
rabbitmq connected: true
...
```
This message indicates that the RabbitMQ connection is connected normally.

## Disconnection & Down
After running and connected normally, you can emulate disconnection & server down by manually shutting down the RabbitMQ server. After the server stops, you should see a message like the one below.
```
rabbitmq connection closed unexpectedly, reason: Exception (320) Reason: "CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'"
rabbitmq connected: false
rabbitmq connected: false
rabbitmq connected: false
rabbitmq connected: false
rabbitmq reconnect failed, err: dial tcp [::1]:5672: connectex: No connection could be made because the target machine actively refused it.
rabbitmq connected: false
rabbitmq connected: false
rabbitmq connected: false
rabbitmq reconnect failed, err: dial tcp [::1]:5672: connectex: No connection could be made because the target machine actively refused it.
rabbitmq connected: false
...
```
This message indicates disconnection from the RabbitMQ connection, and you should set up an alert for this event. 

## Reconnection
After disconnection, you can emulate reconnection by running the RabbitMQ server again. After the server is up and ready, and the reconnection mechanism is successful, you should see a message like the one below.
```
rabbitmq connect success
rabbitmq channels register success
rabbitmq reconnect success
rabbitmq connected: true
rabbitmq connected: true
rabbitmq connected: true
rabbitmq connected: true
rabbitmq connected: true
rabbitmq connected: true
```

# Conclusion

In conclusion, the article demonstrates a robust strategy for handling RabbitMQ reconnections in Go, ensuring reliable communication in distributed systems. By implementing a reconnection loop and monitoring the connection status, the application can gracefully recover from network disruptions or broker restarts. This approach enhances the resilience of the system, ensuring that important data is not lost due to temporary connection issues. Using the amqp091-go library, developers can easily implement these reconnection mechanisms, improving the overall reliability of their RabbitMQ-based applications.

# External Link
- [Full Code Example](https://github.com/dikac/go-rabbitmq-reconnect)
- [amqp091-go Repo](https://github.com/rabbitmq/amqp091-go)