Getting Started with mqutils
mqutils is a production-ready Go message queue abstraction library that provides unified interfaces for 6 major messaging systems. This guide will help you get up and running quickly.
Upgrading from v1? See the UPGRADING.md guide — v2 changed the import paths and the handler contract.
Installation
Add the core library and the backend module(s) you need to your Go project:
1
2
3
4
5
6
7
8
9
| go get go.digitalxero.dev/mqutils/v2
# One or more backends
go get go.digitalxero.dev/mq-amqp/v2
go get go.digitalxero.dev/mq-kafka/v2
go get go.digitalxero.dev/mq-nats/v2
go get go.digitalxero.dev/mq-aws/v2
go get go.digitalxero.dev/mq-gcp/v2
go get go.digitalxero.dev/mq-redis/v2
|
Basic Usage
Creating a Consumer
The simplest way to start using mqutils is with the factory function and a viper configuration:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| package main
import (
"context"
"log"
"github.com/spf13/viper"
"go.digitalxero.dev/mqutils/v2"
"go.digitalxero.dev/mqutils/v2/types"
_ "go.digitalxero.dev/mq-amqp/v2" // register the amqp:// scheme
)
func main() {
// Register a message handler: return nil to acknowledge,
// an error to reject (triggering retry/dead-letter handling)
types.RegisterHandler("process", func(ctx context.Context, msg types.Message) error {
log.Printf("Received: %s", string(msg.Body()))
return nil
})
// Configure the consumer; the URL scheme selects the backend
config := viper.New()
config.Set("url", "amqp://localhost:5672/")
config.Set("queue", "myqueue")
config.Set("handler", "process")
ctx := context.Background()
consumer, err := mqutils.NewConsumer(ctx, config)
if err != nil {
log.Fatal(err)
}
// Start consuming messages (blocks until the context is canceled)
if err := consumer.Run(ctx); err != nil {
log.Fatal(err)
}
}
|
The Typed Builder
If you prefer not to use viper, the typed builder covers the configuration shared by every backend:
1
2
3
4
5
6
7
8
9
| consumer, err := mqutils.NewConsumerBuilder().
WithURL("amqp://localhost:5672/").
WithDestination("myqueue").
WithMaxRetries(3).
WithHandler(func(ctx context.Context, msg types.Message) error {
log.Printf("Received: %s", string(msg.Body()))
return nil
}).
Build(ctx)
|
URL Schemes
mqutils automatically detects the message queue system from the URL scheme. Consumers and producers register identical scheme sets:
| System | URL Schemes | Example |
|---|
| AMQP/RabbitMQ | amqp://, amqps:// | amqp://localhost:5672/ |
| Kafka | kafka://, kafkas:// | kafka://localhost:9092 |
| NATS | nats://, natss://, tls://, jetstream:// | nats://localhost:4222 |
| AWS SQS | sqs://, sqss:// | sqs://region/queue-name |
| GCP Pub/Sub | pubsub://, pubsubs://, gcp:// | pubsub://project-id/topic?subscription=sub |
| Redis | redis://, rediss://, redisstream://, redisstreams:// | redis://localhost:6379/mychannel |
Every backend also accepts the canonical config keys destination, max_retries, and consumer_group alongside its native keys.
Core Concepts
Messages
All messages in mqutils implement the types.Message interface. The most commonly used methods:
1
2
3
4
5
6
7
8
9
10
11
12
13
| type Message interface {
MessageId() string // Unique message identifier
CorrelationId() string // For request-response patterns
ReplyTo() string // Reply destination
Exchange() string // AMQP exchange (empty for other systems)
RoutingKey() string // Message routing key
Headers() map[string]interface{} // Message headers
Body() []byte // Message payload
Publisher() Publisher // Access to publisher for replies
Ack() error // Optional manual acknowledgment
Nack() error // Optional manual rejection
// ... additional metadata methods
}
|
Message Handlers
Handlers return an error: a nil return acknowledges the message, a non-nil return rejects it. The consumer runtime settles the message for you.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| // Simple handler
types.RegisterHandler("process", func(ctx context.Context, msg types.Message) error {
// Process the message
log.Printf("Processing: %s", string(msg.Body()))
return nil
})
// Handler with reply
types.RegisterHandler("request", func(ctx context.Context, msg types.Message) error {
// Process request and send reply
response := []byte("processed")
if msg.ReplyTo() != "" {
return msg.Publisher().Publish(ctx,
msg.CorrelationId(), // correlation ID
"", // exchange (empty for default)
msg.ReplyTo(), // routing key (reply destination)
"text/plain", // content type
response) // body
}
return nil
})
|
Batch Processing
Batch processing works on every backend. A nil return settles the whole batch; an error rejects every message the handler did not settle itself.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| // Register a batch handler globally
types.RegisterBatchHandler("bulk-process", func(ctx context.Context, msgs []types.Message) error {
log.Printf("Processing batch of %d messages", len(msgs))
for _, msg := range msgs {
// Process each message in the batch
log.Printf("Message: %s", string(msg.Body()))
}
return nil
})
// Configure consumer with batch processing
config := viper.New()
config.Set("url", "amqp://localhost:5672/")
config.Set("queue", "myqueue")
config.Set("handler", "bulk-process")
config.Set("enable_batch_processing", true)
config.Set("batch_size", 50)
config.Set("batch_timeout", "1s") // duration string
consumer, err := mqutils.NewConsumer(ctx, config)
|
Health Monitoring
All consumers implement types.HealthChecker:
1
2
3
4
5
6
7
8
9
10
11
12
| // Check health status
health, err := consumer.HealthCheck(context.Background())
if err != nil {
log.Printf("Health check failed: %v", err)
return
}
if health.Status() == types.HealthStatusHealthy {
log.Println("Consumer is healthy")
} else {
log.Printf("Consumer health: %s - %s", health.Status(), health.Message())
}
|
Publishing Messages
Create a producer with the factory function (or mqutils.NewProducerBuilder()):
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
| config := viper.New()
config.Set("url", "amqp://localhost:5672/")
producer, err := mqutils.NewProducer(context.Background(), config)
if err != nil {
log.Fatal(err)
}
if err := producer.Start(context.Background()); err != nil {
log.Fatal(err)
}
// Simple publish
err = producer.Publish(
context.Background(),
"req-123", // correlation ID
"", // exchange (empty for default)
"destination", // routing key/destination
"text/plain", // content type
[]byte("Hello, World!"), // body
)
// Publish with message builder
import "go.digitalxero.dev/mq-amqp/v2"
message := amqp.NewMessageBuilder().
WithCorrelationId("req-123").
WithContentType("application/json").
WithHeaders(map[string]interface{}{
"priority": 1,
}).
WithBody([]byte(`{"event": "user.created", "id": 123}`)).
Build()
err = producer.PublishMsg(context.Background(), "events", "user.created", message)
|
Configuration Examples
AMQP/RabbitMQ
1
2
3
4
5
6
7
| config := viper.New()
config.Set("url", "amqp://user:pass@localhost:5672/")
config.Set("queue", "queue")
config.Set("exchange", "events")
config.Set("routing_key", "user.created")
config.Set("handler", "process")
consumer, err := mqutils.NewConsumer(ctx, config)
|
Kafka
1
2
3
4
5
6
| config := viper.New()
config.Set("url", "kafka://localhost:9092")
config.Set("topic", "mytopic")
config.Set("consumer_group", "my-service")
config.Set("handler", "process")
consumer, err := mqutils.NewConsumer(ctx, config)
|
AWS SQS
1
2
3
4
| config := viper.New()
config.Set("url", "sqs://us-east-1/my-queue") // or my-queue.fifo
config.Set("handler", "process")
consumer, err := mqutils.NewConsumer(ctx, config)
|
Error Handling
mqutils provides structured error handling driven by the handler’s return value:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| types.RegisterHandler("process", func(ctx context.Context, msg types.Message) error {
// Your processing logic here
if err := processMessage(ctx, msg); err != nil {
// Return error to trigger message retry/dead letter
return fmt.Errorf("processing failed: %w", err)
}
return nil // Message will be acknowledged automatically
})
// Manual settlement still works when you need to settle mid-handler.
// The runtime treats the duplicate settlement as success.
types.RegisterHandler("manual", func(ctx context.Context, msg types.Message) error {
if err := processMessage(ctx, msg); err != nil {
// Explicitly nack the message
return msg.Nack()
}
// Explicitly ack the message
return msg.Ack()
})
|
Graceful Shutdown
Always implement graceful shutdown:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
| func main() {
// Register handler first
types.RegisterHandler("process", func(ctx context.Context, msg types.Message) error {
log.Printf("Processing: %s", string(msg.Body()))
return nil
})
// Enable graceful shutdown in configuration
config := viper.New()
config.Set("url", "amqp://localhost:5672/")
config.Set("queue", "myqueue")
config.Set("handler", "process")
config.Set("enable_graceful_shutdown", true)
config.Set("graceful_shutdown_timeout", 30) // seconds
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
consumer, err := mqutils.NewConsumer(ctx, config)
if err != nil {
log.Fatal(err)
}
go func() {
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
<-sigChan
log.Println("Shutdown signal received, stopping consumer...")
cancel()
}()
// Run will drain in-flight messages and return when the context is canceled
if err := consumer.Run(ctx); err != nil {
log.Printf("Consumer error: %v", err)
}
}
|
Note: if the broker connection is lost and auto_reconnect is disabled, Run returns an error. Set auto_reconnect to true to reconnect with backoff instead.
Next Steps