Go · v2 · MIT

One consumer API.
Six brokers.

mqutils gives Go services a single consumer and producer interface over RabbitMQ, Kafka, NATS, SQS, Pub/Sub and Redis. The URL scheme picks the broker. Your handler stays put.

$ go get go.digitalxero.dev/mqutils/v2

$ go get go.digitalxero.dev/mq-kafka/v2 # one module per broker

RabbitMQ Streams: Go 1.25+ and module setup

Get started Read the docs

go 1.24+ license MIT backends 7 handlers return error

import _ "go.digitalxero.dev/mq-kafka/v2" // registers the scheme

consumer, err := mqutils.NewConsumerBuilder().
    WithURL("kafka://localhost:9092").
    WithDestination("user-events").
    WithConsumerGroup("user-service").
    WithMaxRetries(3).
    WithHandler(func(ctx context.Context, msg types.Message) error {
        return process(msg.Body()) // nil acks · error retries
    }).
    Build(ctx)
if err != nil {
    log.Fatal(err)
}
if err := consumer.Run(ctx); err != nil {
    log.Printf("consumer stopped: %v", err)
}

changed: 2 lineshandler: unchangedsettlement: runtime

Backends

What each broker actually guarantees

Same interface, honest semantics. Retry budgets and dead-lettering apply only where the broker can redeliver. NATS Core and Redis Pub/Sub are at-most-once and will not retry.

Backend	Schemes	Delivery	On retry budget exhausted	Notable
AMQP / RabbitMQ	amqp:// amqps://	at-least-once	Dead-letter exchange; native delayed retry on RabbitMQ ≥ 4.3	Channel pooling, publisher confirms, blocked-connection queueing
Apache Kafka	kafka:// kafkas://	at-least-once	Retry topic → dead-letter topic	TLS/SASL, consumer groups, atomic producer transactions
NATS JetStream	jetstream://	at-least-once	Dropped after the budget	Streams, durable consumers, explicit acks
NATS Core	nats:// natss:// tls://	at-most-once	no redelivery	Queue groups, wildcards, blocking backpressure
AWS SQS	sqs:// sqss://	at-least-once	Dead-letter queue via RedrivePolicy	Standard and FIFO, visibility timeout, long polling
GCP Pub/Sub	pubsub:// pubsubs:// gcp://	at-least-once	DeadLetterPolicy (needs an IAM grant)	Ordering keys, publish batching
Redis Streams	redisstream:// redisstreams://	at-least-once	Dropped after the budget	Consumer groups, XAutoClaim recovery (Redis ≥ 6.2), trimming
Redis Pub/Sub	redis:// rediss://	at-most-once	no redelivery	Channels and patterns
RabbitMQ Streams	rabbitmq-stream:// rabbitmq-stream+tls://	at-least-once	Confirmed dead-letter stream, explicit discard, or stop without advancing	Super streams, exact filtering, durable offsets, outbox/inbox deduplication

Handler contract

Return a value. The runtime settles the message.

Return nil to acknowledge or an error to retry. The shared consumer runtime acknowledges, retries with backoff, and dead-letters according to the broker you chose.

return nil

Acknowledged

The message is settled with the broker and removed from the queue or committed as an offset.

return err

Retried

Rejected and redelivered with the broker’s retry mechanism, counted against max_retries.

attempts > max_retries

Dead-lettered or dropped

Sent to the DLX, DLT, DLQ or DeadLetterPolicy target when the broker has one. Otherwise dropped, and logged.

Runtime features

The operational parts you would otherwise write per broker

destination: One canonical key for queue, topic, subject, or stream. Native keys still work.
max_retries: A retry budget that every backend honors, dead-lettering where the broker supports it.
consumer_group: Kafka groups, NATS queue groups, JetStream durables and Redis consumer groups behind one key.
batch_size · batch_timeout: Batch handlers on every backend, with a duration string like "250ms".

graceful shutdown: In-flight messages drain on context cancellation on every backend.
health checks: Interface-based health reporting with transport-specific diagnostics.
blocked connections: When RabbitMQ raises a memory or disk alarm, publishes queue in order until the broker unblocks.
typed builders: Configure consumers and producers in Go without viper, or keep viper if you already use it.