CloudAMQP with Go

We recommend amqp-client.go. It reconnects automatically, recovers queues and consumers after a reconnect, and has no dependencies.

go mod init example.com/cloudamqp-example
go get github.com/cloudamqp/amqp-client.go

Publish and consume

Save the publisher as publisher.go:

package main

import (
	"context"
	"log"

	amqp "github.com/cloudamqp/amqp-client.go"
)

// runPublisher publishes a persistent message and returns once the broker has confirmed it.
func runPublisher(ctx context.Context, c *amqp.Client, queue, message string) error {
	q, err := c.Queue(ctx, queue, nil)
	if err != nil {
		return err
	}
	if err := q.Publish(ctx, amqp.Message{Body: message}); err != nil {
		return err
	}
	log.Printf(" [x] Sent '%s'", message)
	return nil
}

Save the consumer as worker.go in the same directory:

package main

import (
	"context"
	"log"

	amqp "github.com/cloudamqp/amqp-client.go"
)

// runWorker calls handler for each message until ctx is done, with a prefetch of 10.
// The message is acked when handler returns nil, and requeued when it returns an error.
func runWorker(ctx context.Context, c *amqp.Client, queue string, handler func(message string) error) error {
	q, err := c.Queue(ctx, queue, nil)
	if err != nil {
		return err
	}
	_, err = q.Subscribe(ctx, func(ctx context.Context, d *amqp.Delivery) error {
		return handler(string(d.Body))
	}, &amqp.SubscribeOptions{Prefetch: 10})
	if err != nil {
		return err
	}
	log.Println(" [*] Waiting for messages. To exit press CTRL+C")
	<-ctx.Done()
	return nil
}

Save the entry point as main.go. It connects using CLOUDAMQP_URL, which you set to the AMQP URL from your instance’s details page in the CloudAMQP console. Without it, the example connects to a local broker.

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"

	amqp "github.com/cloudamqp/amqp-client.go"
)

func main() {
	if len(os.Args) != 2 || (os.Args[1] != "publish" && os.Args[1] != "worker") {
		fmt.Fprintln(os.Stderr, "Usage: go run . publish|worker")
		os.Exit(2)
	}

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
	defer stop()

	c, err := connect(ctx)
	if err != nil {
		log.Fatal(err)
	}
	defer c.Close()

	if os.Args[1] == "publish" {
		err = runPublisher(ctx, c, "hello", "Hello CloudAMQP!")
	} else {
		err = runWorker(ctx, c, "hello", func(message string) error {
			log.Printf(" [x] Received '%s'", message)
			return nil
		})
	}
	if err != nil {
		log.Fatal(err)
	}
}

// connect connects to CLOUDAMQP_URL, and reconnects if the connection is lost.
func connect(ctx context.Context) (*amqp.Client, error) {
	url := os.Getenv("CLOUDAMQP_URL")
	if url == "" {
		url = "amqp://guest:guest@localhost"
	}
	return amqp.NewClient(ctx, url, nil)
}

Run go run . worker in one terminal and go run . publish in another.

A runnable example is in go/amqp-client in cloudamqp/examples.

Good to know

Connect with the URL. amqp.NewClient(ctx, url, nil), where the URL is the one on your instance’s details page in the CloudAMQP console. With amqps:// the server certificate is verified.
Messages are safe by default. Publish sends persistent messages and returns once the broker has confirmed them.
Messages are acked when your handler returns nil. If it returns an error or panics, the message is requeued. Make handlers idempotent, because a message can be delivered again after a reconnect.
Set a prefetch limit with SubscribeOptions.Prefetch, so a consumer doesn’t buffer more messages than it can handle.

See the API reference for codecs, RPC and the low-level API.

Other Go libraries

amqp091-go, maintained by the RabbitMQ team, without automatic reconnection

← Clojure

Rust →