rabbitmq

English | 中文

A RabbitMQ client built on amqp091-go for go-zero, with built-in publisher confirms, configurable consume concurrency, processing timeouts, distributed tracing, Prometheus metrics, automatic reconnection, and graceful shutdown.

Features

📤 Sender / Listener — Sender publishes to an Exchange, Listener consumes from a Queue
✅ Reliable delivery — publisher confirms + mandatory + persistent delivery: Send returning nil means the broker has accepted the message
🔀 Configurable concurrency — per queue, choose strict ordering (Concurrency=1) or parallel processing (Concurrency=N)
⏱️ Processing timeout — a deadline per message, optionally injected by the host framework
🔗 Distributed tracing — OpenTelemetry integrated; the producer injects a Span, the consumer extracts it and continues the trace
📊 Observability — 20 built-in Prometheus metrics, exposed through go-zero's /metrics
🔄 Auto-reconnect — exponential backoff after a connection/channel closes, interruptible by a shutdown signal
🛡️ Graceful shutdown — stop taking new messages first, then drain in-flight ones, all bounded by GracefulTimeout
🧩 Unified middleware — consuming and publishing share one Middleware abstraction; business middleware can be appended
🔐 TLS — one-way and mutual TLS supported
🏗️ Admin — declare exchanges, queues, and bindings

Installation

bash

go get github.com/lerity-yao/czt-contrib/mq/rabbitmq

Service Generation

Similar to goctl, you can generate server-side code with a single command using cztctl (a customized fork of goctl), which also supports custom code generation templates.

See cztctl for details.

Quick Start

Consumer (Listener)

package main

import (
	"context"
	"time"

	"github.com/lerity-yao/czt-contrib/mq/rabbitmq"
	"github.com/zeromicro/go-zero/core/logx"
)

func main() {
	conf := rabbitmq.RabbitListenerConf{
		RabbitConf: rabbitmq.RabbitConf{
			Username: "guest",
			Password: "guest",
			Host:     "127.0.0.1",
			Port:     5672,
		},
		ListenerQueues: []rabbitmq.ConsumerConf{
			{Name: "queue.demo", Concurrency: 5, Timeout: 3000},
		},
		ChannelQos: rabbitmq.ChannelQosConf{PrefetchCount: 10},
	}

	listener := rabbitmq.MustNewListener(conf, rabbitmq.HandlerFunc(
		func(ctx context.Context, meta rabbitmq.Meta, message []byte) error {
			logx.WithContext(ctx).Infof("received from %s: %s", meta.Queue, message)
			return nil
		}))

	defer listener.Stop()
	listener.Start() // blocks until Stop is called
}

Producer (Sender)

sender := rabbitmq.MustNewSender(rabbitmq.RabbitSenderConf{
	RabbitConf: rabbitmq.RabbitConf{
		Username: "guest",
		Password: "guest",
		Host:     "127.0.0.1",
		Port:     5672,
	},
	ContentType: "application/json",
})
defer sender.Close()

if err := sender.Send(ctx, "exchange.demo", "route.demo", []byte(`{"id":1}`)); err != nil {
	// only nil means the broker has accepted the message
	logx.WithContext(ctx).Errorf("send failed: %v", err)
}

Configuration

RabbitConf (base connection config)

Field	Type	Default	Description
`Username`	string	—	Username, required
`Password`	string	—	Password, required
`Host`	string	—	Host address, required
`Port`	int	—	Port, required, must be within 1–65535
`VHost`	string	empty	Virtual host
`Heartbeat`	int64	`30`	Heartbeat interval in seconds; detects dead links and triggers reconnection
`TLS`	TLSConf	disabled	TLS config

TLSConf

Field	Type	Default	Description
`Enable`	bool	`false`	When enabled, the dial scheme switches from `amqp` to `amqps`
`CACertFile`	string	empty	CA certificate path; the system root pool is used when empty
`ClientCertFile`	string	empty	Client certificate path, for mutual authentication
`ClientKeyFile`	string	empty	Client private key path; must be set together with `ClientCertFile` or both left empty
`InsecureSkipVerify`	bool	`false`	Skip server certificate verification; only for test environments with self-signed certificates

RabbitSenderConf (producer config)

Embeds RabbitConf, plus:

Field	Type	Default	Description
`ContentType`	string	`text/plain`	MIME type of the published message
`PoolSize`	int	`10`	Confirm channel pool size; publish throughput is roughly `PoolSize / network RTT`
`PublishTimeout`	int64	`5000`	Overall timeout of one `Send` in milliseconds, covering channel checkout queuing, confirm waiting, and all retries
`MaxRetry`	int	`3`	Retry count for retryable failures; `0` disables retrying
`GracefulTimeout`	int64	`10000`	Graceful shutdown timeout in milliseconds: the upper bound `Close` waits for in-flight publishes to be confirmed by the broker

RabbitListenerConf (consumer config)

Embeds RabbitConf, plus:

Field	Type	Default	Description
`ListenerQueues`	[]ConsumerConf	—	Queues to consume, required and non-empty; each queue owns a dedicated channel
`ChannelQos`	ChannelQosConf	—	Consumer channel QoS, applied to every queue
`GracefulTimeout`	int64	`10000`	Graceful wait timeout in milliseconds: the upper bound for draining in-flight messages on reconnect/stop

ConsumerConf (per-queue consume config)

Field	Type	Default	Description
`Name`	string	—	Queue name, required and must be unique
`AutoAck`	bool	`false`	When `true` the broker acks on delivery and the framework performs no acknowledgement
`Exclusive`	bool	`false`	Exclusive queue access
`NoWait`	bool	`false`	When `true`, do not wait for the broker's response — you cannot tell whether the consumer was created
`Concurrency`	int	`1`	Consume concurrency per queue, capped at 256
`Timeout`	int64	`0`	Per-message processing timeout in milliseconds; `0` means unlimited. Overridden by `WithConsumeTimeout` when that option is passed

The two modes of Concurrency:

equals 1: strict ordering — messages of a queue are processed serially in broker delivery order
greater than 1: parallel — that many worker goroutines process in parallel, and ordering within the queue no longer holds

Constraint: Concurrency <= ChannelQos.PrefetchCount, otherwise the surplus workers would never receive a message. Validate enforces this.

ChannelQosConf (channel QoS config)

Field	Type	Default	Description
`PrefetchCount`	int	`5`	Cap on unacknowledged messages per queue; the broker stops delivering once reached. Capped at 65535
`PrefetchSize`	int	`0`	Not implemented by RabbitMQ; must be 0 or `Validate` fails

PrefetchCount is not a concurrency setting. It is a flow-control parameter (driven by message size and network RTT), while concurrency is set by Concurrency (driven by downstream capacity). They serve different purposes.

PrefetchCount is per queue, not per Listener. QoS is set per channel and each queue owns a dedicated channel, so with N queues the in-flight ceiling is PrefetchCount × N. Size memory as PrefetchCount × queue count × message size, and lower the count when consuming many queues.

ExchangeConf / QueueConf (declaration config)

ExchangeConf field	Type	Default	Description
`ExchangeName`	string	—	Exchange name
`Type`	string	—	`direct` / `fanout` / `topic` / `headers`
`Durable`	bool	`true`	Durable
`AutoDelete`	bool	`false`	Delete when no longer bound
`Internal`	bool	`false`	Internal exchange
`NoWait`	bool	`false`	Do not wait for the broker's response
`Queues`	[]QueueConf	—	Associated queues

QueueConf field	Type	Default	Description
`Name`	string	—	Queue name
`Durable`	bool	`true`	Durable
`AutoDelete`	bool	`false`	Delete when no consumer remains
`Exclusive`	bool	`false`	Exclusive
`NoWait`	bool	`false`	Do not wait for the broker's response

The three validation rules

Validate applies exactly one rule per field, so the behaviour is unambiguous:

Rule	Action	Applies to	Examples
A	Return an error (fail fast)	A required field is missing, or the value cannot possibly be correct	empty `Username`/`Password`/`Host`, out-of-range `Port`, duplicate queue names, `PrefetchSize != 0`
B	Fall back to the default and log one line	An invalid value that has a safe default	`Heartbeat <= 0`, `Concurrency <= 0`, `PoolSize <= 0`
C	Return an error (limit guard)	Beyond a protocol or resource limit	`PrefetchCount > 65535`, `Concurrency > 256`, `Concurrency > PrefetchCount`

Every new field has a default, so existing configurations in your config center need no changes to upgrade.

Timeouts

Level	Setting	Default	Mechanism
Per message	`ConsumerConf.Timeout`	`0` (unlimited)	`timeoutMiddleware`
Host injection (consume)	`WithConsumeTimeout(d)`	—	Overrides `Timeout` for every queue
One publish	`RabbitSenderConf.PublishTimeout`	`5000`	`timeoutMiddleware`
Host injection (publish)	`WithPublishTimeout(d)`	—	Overrides `PublishTimeout`
Graceful shutdown	`RabbitListenerConf.GracefulTimeout`	`10000`	Wait bound in `Stop` / `reconnect`
Graceful shutdown (publish)	`RabbitSenderConf.GracefulTimeout`	`10000`	Bound `Close` waits for in-flight publishes to be confirmed

Precedence: a deadline already on the caller's ctx > option (explicitly passed by the host) > conf field > default. An option of 0 or a negative value counts as unset.

The key semantic: a configured value is a backstop that applies only when the caller supplied no deadline. timeoutMiddleware first checks whether the incoming ctx already carries a deadline and passes it through untouched if so. This differs from wrapping context.WithTimeout(ctx, d) unconditionally: that takes the earlier of the two, silently truncating a longer deadline the caller explicitly asked for, which turns the configured value from a backstop into a hard ceiling.

⚠️ Deadlines are cooperative. The handler must pass ctx to every downstream blocking call, or the timeout has no effect — time.Sleep or any call that ignores ctx will occupy the worker until it finishes on its own.

On the consumer side the message ctx is derived by the Listener from context.Background() (never from the generation ctx, or a reconnect would cancel in-flight messages) and therefore carries no deadline, so Timeout and WithConsumeTimeout always apply. Send, by contrast, receives its ctx from the caller; if that already carries a deadline (inside a go-zero rest handler, for instance), PublishTimeout does not intervene.

When hosted by go-zero, inject the host's RestConf.Timeout explicitly in the handler:

return rabbitmq.MustNewListener(
	svcCtx.Config.GDemoARabbitmqConf,
	rabbitmq.HandlerFunc(handler),
	rabbitmq.WithConsumeTimeout(time.Duration(svcCtx.Config.Timeout)*time.Millisecond),
)

Publish semantics

Send returning nil means the broker has accepted the message, backed by three guarantees:

Publisher confirms — every publish waits for the broker's ack/nack instead of returning once written to the socket
mandatory=true — a message whose routeKey matches no queue is returned and reported as an error, instead of being silently dropped by the broker
DeliveryMode=Persistent — avoids the half-durable trap of "durable queue + transient message"

Error classification

Failures can be classified with errors.Is:

Sentinel error	Meaning	Retried	Suggested handling
`ErrPublishUnroutable`	No queue is bound to that routeKey; the message was returned	No	A configuration error — alert rather than republish
`ErrPublishNacked`	The broker explicitly refused (disk alarm, internal error, …)	Yes	Already retried with backoff inside the framework
`ErrPublishTimeout`	Confirm wait timed out; whether the broker got the message is unknown	No	Let the business decide how to compensate
`ErrPublishUnknown`	The channel closed before the confirm arrived (usually a dropped connection); the outcome is unknown. The broker's close reason (e.g. `404 NOT_FOUND`) is appended to the error when available	No	Same as above; read the close reason first, then investigate network and connection stability
`ErrSenderClosed`	The sender is already closed	No	A publish after shutdown — a call-ordering issue

The dividing line is whether the outcome is known. ErrPublishNacked and ErrPublishUnroutable both carry a complete response from the broker, so the outcome is settled. ErrPublishTimeout and ErrPublishUnknown never got a confirm — the message may or may not have landed, and the framework deliberately does not retry either of them: republishing would duplicate the message behind the caller's back, and only the business knows whether that is acceptable.

switch {
case errors.Is(err, rabbitmq.ErrPublishUnroutable):
	// wrong routeKey or missing binding
case errors.Is(err, rabbitmq.ErrPublishTimeout), errors.Is(err, rabbitmq.ErrPublishUnknown):
	// the message may already be delivered; republishing may duplicate it
}

Retrying

Retryable failures are retried within the PublishTimeout budget with exponential backoff starting at 50ms, up to MaxRetry times. Retrying stops as soon as the budget is exhausted, returning the last error as-is.

Only failures whose outcome is certain are retried: a nack (the broker refused it outright) and errors raised before the frame ever left (no channel could be borrowed, or the write itself was rejected). Anything that leaves the outcome unknown is handed back to the caller.

⚠️ Retrying inevitably duplicates messages. At-least-once is this module's delivery guarantee, so consumers must be idempotent — give every message a business-unique key and deduplicate before processing.

The channel pool

amqp091-go's Channel is not thread-safe, and NotifyReturn carries no deliveryTag, so a returned message cannot be attributed to the right one among several in flight on the same channel. Concurrent publishing therefore goes through a channel pool where each channel handles a single in-flight message at a time, with concurrency set by PoolSize.

A channel that received a complete confirmation (ack, nack, or return) is in a clean state and is reused directly. A confirm timeout or a connection/channel failure leaves a dangling unconfirmed message in an unknown state, so such channels are always discarded and rebuilt.

Consume semantics

Acknowledgement decisions

The framework does not retry (keeping the established "a failure is a failure" stance). Implement retrying inside the handler if you need it (delay queue, persisting to a DB, …).

Situation	Action	Rationale
Processed successfully	`Ack`	—
The handler returned an error	`Ack`	Redelivering as-is would only fail again
Processing timed out	`Ack`	Same as above
The handler or a middleware panicked	`Ack`	The panic is turned into an error in place: recovery covers the business handler, the consumer covers a middleware panic escaping the chain
Intercepted during shutdown (never executed)	`Reject(requeue=true)`	The message was not processed yet — return it so another instance takes over
`AutoAck=true`	Nothing	Acknowledgement is owned by the broker

Concurrency and ordering

With Concurrency=N, N worker goroutines behind the queue's channel take messages from the same delivery stream. In ordered mode (N=1) messages are processed serially in broker delivery order.

Throughput ceiling per queue: 1 / per-message processing time in ordered mode, and N / per-message processing time in parallel mode, bounded by PrefetchCount flow control.

Queue layout: one Listener or several

A Listener owns one connection, and three settings apply to all of its queues at once: ChannelQos, GracefulTimeout, and WithConsumeTimeout. Reconnection is also connection-scoped — when the connection drops, every queue's consumer generation is cancelled and rebuilt together.

Put queues in one Listener when their flow-control needs and processing latencies are of the same order. Split them into separate Listeners when:

Prefetch needs diverge — ChannelQos cannot be varied per queue, so a queue of large messages and a queue of small ones have to share one PrefetchCount
Processing latencies diverge — GracefulTimeout and WithConsumeTimeout are shared, so one slow queue dictates the shutdown budget for all the fast ones
Fault isolation matters — a single connection failure rebuilds every queue, so an unrelated queue's blip stalls a critical one

Each queue keeps its own channel, Concurrency and Timeout regardless, so splitting buys isolation at the cost of one extra TCP connection per Listener.

Middleware

Consuming and publishing share one abstraction — Meta describes the message coordinates (the consumer side fills Queue only, the publisher side Exchange and RouteKey), so there is no need to maintain two isomorphic sets of types and chain builders:

type Handler func(ctx context.Context, meta Meta, body []byte) error
type Middleware func(next Handler) Handler

func Chain(middlewares ...Middleware) Middleware

In Chain, earlier middleware sits further out: it sees the request first and the result last.

Built-in chain order:

Listener: metrics → logging → consumerTrace → timeout → business middleware → recovery → business handler
Sender: metrics → logging → producerTrace → timeout → business middleware → recovery → publish (with retries)

Recovery sits innermost: were it outermost, a panic would pass through metrics and logging, so panicking messages would never be counted in consume_total.

Appending business middleware

func auditMiddleware(next rabbitmq.Handler) rabbitmq.Handler {
	return func(ctx context.Context, meta rabbitmq.Meta, body []byte) error {
		start := time.Now()
		err := next(ctx, meta, body)
		logx.WithContext(ctx).Infow("audit",
			logx.Field("queue", meta.Queue),
			logx.Field("cost", time.Since(start).String()),
			logx.Field("err", err))
		return err
	}
}

listener := rabbitmq.MustNewListener(conf, handler,
	rabbitmq.WithConsumeMiddlewares(auditMiddleware))

sender := rabbitmq.MustNewSender(senderConf,
	rabbitmq.WithPublishMiddlewares(auditMiddleware))

Business middleware is appended after the built-in chain (closer to the business handler) and is therefore bounded by the processing timeout. On the publishing side it sits outside the retry loop, so it runs exactly once per Send rather than once per attempt.

Only appending is offered, never replacing: recovery and metrics are framework-level guarantees, and allowing replacement would let callers disable the panic safety net and metric collection without realising it.

Distributed tracing

The trace context travels in the AMQP header table, following the OpenTelemetry messaging convention. The producer injects it into Meta.Headers, which ends up as amqp.Publishing.Headers; the consumer extracts it from amqp.Delivery.Headers and continues the trace, transparently to the business code.

The message body is never touched. Three consequences follow:

The body need not be JSON. Protobuf, plain text, a bare ID — anything works, since nothing is wrapped around it.
Messages from other producers are consumable. A message published by a Java service, or by hand from the RabbitMQ management UI, carries no trace header. That only means there is no upstream trace to continue; the message is still delivered to the handler.
Traces cross language boundaries. Any producer whose OpenTelemetry SDK writes traceparent into the AMQP headers — regardless of language — has its trace continued here, and vice versa.

The message handed to the handler is exactly the bytes the producer published, and ctx already carries the trace context — logx.WithContext(ctx) is enough to correlate logs with traces.

Metrics

Exposed through go-zero's /metrics.

Sender

Metric	Type	Labels	Purpose
`rabbitmq_sender_send_total`	counter	exchange, route_key, status	Publishes, classified by outcome
`rabbitmq_sender_send_duration_ms`	histogram	exchange, route_key	End-to-end duration of one `Send`
`rabbitmq_sender_send_size_bytes`	histogram	exchange, route_key	Payload size
`rabbitmq_sender_confirm_duration_ms`	histogram	exchange, route_key	Time spent waiting for the broker's confirm
`rabbitmq_sender_pool_wait_duration_ms`	histogram	endpoint, vhost	Channel checkout queuing time
`rabbitmq_sender_pool_in_use`	gauge	endpoint, vhost	Pool occupancy; tells whether `PoolSize` is too small
`rabbitmq_sender_retry_total`	counter	exchange, route_key	Retry count
`rabbitmq_sender_reconnect_total`	counter	endpoint, vhost	Reconnections
`rabbitmq_sender_disconnect_total`	counter	endpoint, vhost	Disconnections

status values: success / fail / timeout / unroutable / nacked / unknown.

Listener

Metric	Type	Labels	Purpose
`rabbitmq_listener_consume_total`	counter	queue, status	Consumptions, classified by outcome
`rabbitmq_listener_consume_duration_ms`	histogram	queue	Per-message processing duration
`rabbitmq_listener_consume_size_bytes`	histogram	queue	Payload size
`rabbitmq_listener_in_flight`	gauge	queue	In-flight messages; tells whether `Concurrency` is too small
`rabbitmq_listener_panic_total`	counter	queue	Handler panics
`rabbitmq_listener_ack_total`	counter	queue, type	Acknowledgements, `type` being `ack` / `reject`
`rabbitmq_listener_reconnect_total`	counter	endpoint, vhost	Reconnections
`rabbitmq_listener_disconnect_total`	counter	endpoint, vhost	Disconnections

status values: ok / fail / timeout / canceled / panic.

endpoint (host:port) and vhost tell apart Senders / Listeners connected to different clusters within one process; vhost falls back to the AMQP default / when unset.

Log prefixes

Situation	Level	Prefix
Intercepted during shutdown (requeued without executing)	Info	`[RABBITMQ_CANCELED]`
Processing/publish timeout	Error	`[RABBITMQ_TIMEOUT]`
Unroutable	Error	`[RABBITMQ_UNROUTABLE]`
Nacked	Error	`[RABBITMQ_NACKED]`
Publish retry	Error	`[RABBITMQ_RETRY]`

Auto-reconnect

The first connection at construction time is attempted only once: configuration errors should surface immediately at startup, while runtime connection failures are healed by reconnection.

The connection's NotifyClose is watched, and after a disconnect the client reconnects with exponential backoff (starting at 2s and growing), with the backoff wait interruptible by a shutdown signal. One Listener connection corresponds to one "generation" of consumer goroutines: on reconnect the old generation is cancelled and drained first (bounded by GracefulTimeout) before a new one is established, which rules out double consumption by two generations.

Graceful shutdown

Listener

Stop() shuts down in three stages, bounded by GracefulTimeout so that it always returns:

Switch to the stopping state; newly arriving messages are all Reject(requeue=true)-ed back to the queue
Cancel the current generation and wait for in-flight messages to finish
Close the channels and the connection

In-flight messages' ctx is not cancelled on shutdown — that would misreport "not finished" as a business failure and leave half-finished intermediate states.

Stop is idempotent; repeated calls do nothing. It is called automatically when hosted by go-zero's ServiceGroup.

Sender

A shutdown hook is registered at construction via proc.AddShutdownListener, so Close runs automatically before the process exits. Close is idempotent (an explicit call and the hook each invoke it once); it cancels an in-progress reconnect backoff, drains the channel pool, and closes the connection.

Admin

Used to declare topology at service startup. Admin is a one-shot tool that neither reconnects nor pools connections — declarations are idempotent and only run at startup.

⚠️ You must Close it, otherwise one AMQP connection leaks. Close is idempotent.

admin, err := rabbitmq.NewAdmin(conf)
if err != nil {
	return err
}
defer admin.Close()

if err = admin.DeclareExchange(rabbitmq.ExchangeConf{
	ExchangeName: "exchange.demo",
	Type:         "direct",
	Durable:      true,
}, nil); err != nil {
	return err
}
if err = admin.DeclareQueue(rabbitmq.QueueConf{Name: "queue.demo", Durable: true}, nil); err != nil {
	return err
}
return admin.Bind("queue.demo", "route.demo", "exchange.demo", false, nil)

go-zero integration example

Consumer side

etc/demoa.yaml:

yaml

Name: demoa
Host: 0.0.0.0
Port: 8888
Timeout: 30000                # injected as the per-message processing bound

GDemoARabbitmqConf:
  Username: guest
  Password: guest
  Host: 127.0.0.1
  Port: 5672
  GracefulTimeout: 10000
  ChannelQos:
    PrefetchCount: 10
  ListenerQueues:
    - Name: queue.demoa
      Concurrency: 5          # parallel; use 1 or omit for strict ordering

internal/config/config.go:

type Config struct {
	rest.RestConf
	GDemoARabbitmqConf rabbitmq.RabbitListenerConf
}

internal/handler/demoA/gdemoahandler.go (generated by cztctl 1.11.0+ as-is):

func GDemoAHandler(svcCtx *svc.ServiceContext) service.Service {
	handler := func(ctx context.Context, meta rabbitmq.Meta, message []byte) error {
		l := demoA.NewGDemoALogic(ctx, meta, svcCtx)
		return l.GDemoA(message)
	}
	return rabbitmq.MustNewListener(
		svcCtx.Config.GDemoARabbitmqConf,
		rabbitmq.HandlerFunc(handler),
		rabbitmq.WithConsumeTimeout(time.Duration(svcCtx.Config.Timeout)*time.Millisecond),
	)
}

meta is injected into the logic constructor rather than passed to the business method, so l.meta.Queue is available from every method — that is what lets one handler serve several queues. When the DSL declares a message body type, the generated handler unmarshals it and the business method receives *types.XxxReq instead of []byte.

main.go:

group := service.NewServiceGroup()
defer group.Stop()

group.Add(demoA.GDemoAHandler(ctx))
group.Start()

Prefer the HandlerFunc closure form: a new Logic instance per message picks up that consumption's ctx (carrying trace context and deadline) instead of reusing the one from construction time.

A complete runnable example lives in example.

Producer side

type ServiceContext struct {
	Config     config.Config
	DemoSender rabbitmq.Sender
}

func NewServiceContext(c config.Config) *ServiceContext {
	return &ServiceContext{
		Config:     c,
		DemoSender: rabbitmq.MustNewSender(c.DemoSenderConf),
	}
}

func (l *SendLogic) Send(req *types.SendReq) error {
	body, err := json.Marshal(req)
	if err != nil {
		return err
	}
	return l.svcCtx.DemoSender.Send(l.ctx, "exchange.demo", "route.demo", body)
}

Upgrading from v0.1.x

No configuration change is required, but the following breaking changes need attention. See the CHANGELOG for the full list.

Change	Impact
`Interceptor` / `SenderInterceptor` / `SenderFunc` / `SenderChain` removed in favour of `Meta` / `Handler` / `Middleware` / `Chain`	Custom interceptors, if any, must be rewritten
`Send` now blocks for the confirm, and errors are classified into five sentinels	Higher latency and lower throughput, compensated by `PoolSize`; callers can branch by error type
Publishing defaults to `Persistent` + `mandatory`	A wrong routeKey now returns an error instead of silently succeeding
`ConsumerConf.NoLocal` removed	No impact (RabbitMQ never implemented the flag; the extra yaml key is ignored)
`ChannelQosConf.Global` removed	No impact (the flag only scopes allowances within one channel, and each channel here has exactly one consumer, so both values behaved identically; `Qos` now always passes `false`)
`RabbitListenerConf.ContentType` removed	No impact (it was a dead field)

Recommended change: inject the host timeout into existing handlers, i.e. add the rabbitmq.WithConsumeTimeout(...) line at the MustNewListener call by hand (and "time" to the imports). cztctl's handler template is served from a remote repository and is out of this module's scope; besides, handler files are skipped if they already exist during generation, so existing services have to add it by hand either way.

⚠️ rest.RestConf.Timeout defaults to only 3000ms. Once injected, message processing is implicitly capped at 3 seconds, so consumers usually need to raise Timeout explicitly in the yaml.

Testing

bash

# unit tests, no broker required
go test -race .

# integration tests, a real RabbitMQ required
docker run -d --rm -p 5672:5672 rabbitmq:3.13-management
go test -tags=integration -race -timeout=300s .

Integration test connection parameters can be overridden by environment variables: RABBITMQ_HOST / RABBITMQ_PORT / RABBITMQ_USER / RABBITMQ_PASS.

Changelog

See CHANGELOG.md.

rabbitmq ​

Features ​

Installation ​

Service Generation ​

Quick Start ​

Consumer (Listener) ​

Producer (Sender) ​

Configuration ​

RabbitConf (base connection config) ​

TLSConf ​

RabbitSenderConf (producer config) ​

RabbitListenerConf (consumer config) ​

ConsumerConf (per-queue consume config) ​

ChannelQosConf (channel QoS config) ​

ExchangeConf / QueueConf (declaration config) ​

The three validation rules ​

Timeouts ​

Publish semantics ​

Error classification ​

Retrying ​

The channel pool ​

Consume semantics ​

Acknowledgement decisions ​

Concurrency and ordering ​

Queue layout: one Listener or several ​

Middleware ​

Appending business middleware ​

Distributed tracing ​

Metrics ​

Sender ​

Listener ​

Log prefixes ​

Auto-reconnect ​

Graceful shutdown ​

Listener ​

Sender ​

Admin ​

go-zero integration example ​

Consumer side ​

Producer side ​

Upgrading from v0.1.x ​

Testing ​

Changelog ​

rabbitmq

Features

Installation

Service Generation

Quick Start

Consumer (Listener)

Producer (Sender)

Configuration

RabbitConf (base connection config)

TLSConf

RabbitSenderConf (producer config)

RabbitListenerConf (consumer config)

ConsumerConf (per-queue consume config)

ChannelQosConf (channel QoS config)

ExchangeConf / QueueConf (declaration config)

The three validation rules

Timeouts

Publish semantics

Error classification

Retrying

The channel pool

Consume semantics

Acknowledgement decisions

Concurrency and ordering

Queue layout: one Listener or several

Middleware

Appending business middleware

Distributed tracing

Metrics

Sender

Listener

Log prefixes

Auto-reconnect

Graceful shutdown

Listener

Sender

Admin

go-zero integration example

Consumer side

Producer side

Upgrading from v0.1.x

Testing

Changelog