What is RabbitMQ?
RabbitMQ is the most widely deployed open-source message broker, implementing the AMQP 0-9-1 protocol. It acts as an intermediary for messages: producers publish to exchanges, exchanges route messages to queues via binding rules, and consumers subscribe to queues to receive and process messages asynchronously.
Unlike pub/sub systems that deliver to topics directly, RabbitMQ's exchange-binding model gives fine-grained control over routing: a single message can be fanned out to dozens of queues, filtered by routing key pattern, or selectively delivered based on message headers.
Common messaging patterns supported:
SetQoS (prefetch) for fair dispatch.
* (one word) and # (zero or more words): e.g. orders.*.paid or logs.#.
Ubuntu/Debian: sudo apt install librabbitmq-dev
AMQP Concepts
Understanding the four core AMQP entities is essential for designing message flows correctly.
| Concept | Description |
|---|---|
| Exchange | Receives messages from producers and routes them to queues. Never stores messages itself. Types: Direct (exact key match), Fanout (broadcast all), Topic (pattern match), Headers (match by header attributes). |
| Queue | Buffer that stores messages until consumed. Can be durable (survives broker restart), exclusive (tied to one connection), or auto-delete (deleted when last consumer disconnects). |
| Binding | Rule linking an exchange to a queue with an optional routing key or header filter. A queue can be bound to multiple exchanges; an exchange can bind to many queues. |
| Consumer | Application that subscribes to a queue. Can operate in auto-ack mode (fire and forget) or manual ack mode where each message must be explicitly acknowledged with Ack or rejected with Nack/Reject. |
Types
Handles
| Type | Description |
|---|---|
| TRMQConn | Opaque connection handle representing a TCP connection to a RabbitMQ server. |
| TRMQChannel | Opaque channel handle. Channels are lightweight multiplexed logical connections within a single TCP connection. Most operations are per-channel. |
TRMQExchangeType
type TRMQExchangeType = (
etDirect = 0, { exact routing key match }
etFanout = 1, { broadcast to all bound queues }
etTopic = 2, { pattern match: 'orders.*', 'logs.#' }
etHeaders = 3 { route by message header attributes }
);
TRMQConnOptions
type TRMQConnOptions = record
Host : string; { RabbitMQ server hostname or IP }
Port : Integer; { default 5672; TLS uses 5671 }
VHost : string; { virtual host, default '/' }
Username : string;
Password : string;
UseTLS : Boolean;
CACertPath : string; { path to CA certificate for TLS }
HeartbeatSec: Integer; { 0 = disabled }
FrameMax : Integer; { max frame size, default 131072 }
ConnTimeout : Integer; { connection timeout in ms }
end;
TRMQMessage
type TRMQMessage = record
Body : string;
RoutingKey : string;
Exchange : string;
ContentType : string; { e.g. 'application/json' }
ContentEncoding: string;
DeliveryTag : Int64; { used with Ack/Nack }
Redelivered : Boolean;
Persistent : Boolean; { delivery mode 2 }
Priority : Integer; { 0-255 }
MessageID : string;
CorrelationID : string;
ReplyTo : string;
Expiration : string; { TTL in milliseconds as string }
Timestamp : Int64;
Headers : array of record Key, Value: string; end;
end;
TRMQPublishOptions
type TRMQPublishOptions = record
Exchange : string;
RoutingKey : string;
Mandatory : Boolean; { return message if no queue matches }
Immediate : Boolean; { return message if no consumer ready }
Persistent : Boolean; { delivery mode 2 — survive restart }
Priority : Integer;
ContentType : string;
MessageID : string;
CorrelationID: string;
ReplyTo : string;
TTLMs : Integer; { 0 = no expiration }
end;
TRMQQueueInfo
type TRMQQueueInfo = record
Name : string;
MessageCount : Int64;
ConsumerCount: Integer;
Durable : Boolean;
AutoDelete : Boolean;
Exclusive : Boolean;
end;
API Reference
Connection
| Function | Description |
|---|---|
| Connect(host, port, vhost, user, pass) | Connect to a RabbitMQ server with simple credentials. Returns a TRMQConn handle ready for use. |
| ConnectEx(opts) | Connect using a TRMQConnOptions record for full control: TLS, heartbeat interval, frame size, and connection timeout. |
| Disconnect(conn) | Send an AMQP connection close and free all resources associated with the handle. |
| IsConnected(conn) | Returns True if the connection is currently active. |
Channels
| Function | Description |
|---|---|
| OpenChannel(conn, channelID) | Open a channel on the connection. channelID is an integer from 1 to 65535; each channel is independent. Returns a TRMQChannel handle. |
| CloseChannel(ch) | Close the channel and release its server-side resources. |
| SetQoS(ch, prefetchCount) | Set the maximum number of unacknowledged messages the broker will deliver to this channel at once. Essential for fair work-queue dispatch among multiple consumers. |
Exchange & Queue
| Function | Description |
|---|---|
| DeclareExchange(ch, name, exType, durable, autoDelete) | Declare an exchange. durable=True persists the exchange across broker restarts. autoDelete=True deletes it when the last queue unbinds. |
| DeleteExchange(ch, name) | Delete a named exchange from the broker. |
| DeclareQueue(ch, name, durable, exclusive, autoDelete) | Declare a queue. Returns the actual queue name (useful when name='' to get a broker-generated unique name). Durable queues survive restarts; exclusive queues are connection-scoped. |
| DeclareQueueEx(ch, name, durable, exclusive, autoDelete, ttlMs, maxLength) | Declare a queue with a message TTL and maximum depth. Messages older than ttlMs or beyond maxLength are discarded or dead-lettered. |
| BindQueue(ch, queue, exchange, routingKey) | Bind a queue to an exchange using the given routing key. For fanout exchanges the key is ignored. |
| UnbindQueue(ch, queue, exchange, routingKey) | Remove an existing binding between a queue and an exchange. |
| DeleteQueue(ch, name) | Delete a queue and discard all its messages. |
| PurgeQueue(ch, name) | Delete all messages from a queue without removing the queue itself. |
| GetQueueInfo(ch, name) | Return a TRMQQueueInfo record with the current message count, consumer count, and queue flags. |
Publishing
| Function | Description |
|---|---|
| Publish(ch, exchange, routingKey, body, persistent) | Publish a message body string. Set persistent=True for delivery mode 2 (message survives a broker restart when stored in a durable queue). |
| PublishEx(ch, body, opts) | Publish with full control via a TRMQPublishOptions record: headers, TTL, priority, content type, correlation ID, mandatory/immediate flags, and more. |
Consuming
| Function | Description |
|---|---|
| Consume(ch, queue, consumerTag, autoAck) | Register a consumer on the queue. consumerTag is a unique string identifier for this subscription. Set autoAck=False for manual acknowledgement to prevent message loss on consumer crash. |
| CancelConsumer(ch, consumerTag) | Deregister a consumer by its tag. The broker stops delivering messages to this consumer immediately. |
| NextMessage(ch, timeoutMs) | Block until a message arrives on any active consumer for this channel, or until timeoutMs elapses. Returns a TRMQMessage; check Body <> '' to detect a real message vs. timeout. |
Acknowledgements
| Function | Description |
|---|---|
| Ack(ch, deliveryTag, multiple) | Acknowledge successful processing. The broker discards the message. Set multiple=True to ack all messages up to and including deliveryTag in a single call. |
| Nack(ch, deliveryTag, multiple, requeue) | Negatively acknowledge. Set requeue=True to return the message to the queue for redelivery, or False to discard it (or send to dead-letter exchange if configured). |
| Reject(ch, deliveryTag, requeue) | Reject a single message. Equivalent to Nack with multiple=False. |
Transactions
| Function | Description |
|---|---|
| TxSelect(ch) | Enable AMQP transaction mode on the channel. Publishes and acks are not committed until TxCommit is called. |
| TxCommit(ch) | Commit all pending publishes and acknowledgements in the current transaction. |
| TxRollback(ch) | Roll back all pending publishes and acknowledgements in the current transaction. |
Publisher Confirms
| Function | Description |
|---|---|
| ConfirmSelect(ch) | Enable publisher confirms on the channel. The broker will send an ACK or NACK for each published message, enabling the producer to detect broker-side failures. |
| WaitConfirms(ch, timeoutMs) | Block until all pending publisher confirms arrive or the timeout elapses. Returns True if all messages were confirmed successfully. |
Info
| Function | Description |
|---|---|
| RMQVersion | Return the version string of the underlying rabbitmq-c C library. |
Code Example
Publisher and consumer pattern: declare a direct exchange, bind a durable work queue, publish one task, then consume and acknowledge it.
uses rabbitmq-c;
var
Conn: TRMQConn;
Ch : TRMQChannel;
Opts: TRMQConnOptions;
Msg : TRMQMessage;
begin
{ Configure connection }
Opts.Host := 'localhost'; Opts.Port := 5672;
Opts.Username := 'guest'; Opts.Password := 'guest';
Opts.VHost := '/';
{ Connect and open channel 1 }
Conn := ConnectEx(Opts);
Ch := OpenChannel(Conn, 1);
{ Topology: exchange + durable queue + binding }
DeclareExchange(Ch, 'tasks', etDirect, True, False);
DeclareQueue (Ch, 'work_queue', True, False, False);
BindQueue (Ch, 'work_queue', 'tasks', 'work_queue');
{ Publish a persistent task message }
Publish(Ch, 'tasks', 'work_queue', 'Hello World', True);
WriteLn('Published: Hello World');
{ Consume with manual ack, prefetch = 1 for fair dispatch }
SetQoS(Ch, 1);
Consume(Ch, 'work_queue', 'consumer1', False);
{ Wait up to 5 seconds for a message }
Msg := NextMessage(Ch, 5000);
if Msg.Body <> '' then
begin
WriteLn('Received: ', Msg.Body);
if Msg.Redelivered then
WriteLn(' (redelivered message)');
Ack(Ch, Msg.DeliveryTag, False);
end;
CloseChannel(Ch);
Disconnect(Conn);
end.
Fanout (pub/sub broadcast)
uses rabbitmq-c;
var Conn := Connect('localhost', 5672, '/', 'guest', 'guest');
var Ch := OpenChannel(Conn, 1);
{ Fanout exchange: routing key is ignored }
DeclareExchange(Ch, 'notifications', etFanout, True, False);
{ Each subscriber creates its own exclusive, auto-delete queue }
var Q1 := DeclareQueue(Ch, '', False, True, True); { broker-named }
BindQueue(Ch, Q1, 'notifications', '');
{ Publish: all bound queues receive a copy }
Publish(Ch, 'notifications', '', '{"event":"user_signup"}', False);
Disconnect(Conn);
Use autoAck=False + Ack/Nack for at-least-once delivery. Enable ConfirmSelect on the publisher channel to detect broker-side failures. For exactly-once semantics combine publisher confirms with AMQP transactions (TxSelect/TxCommit).
Install
Requires librabbitmq-dev
Package Info
Notes
TLS connections use port 5671. Set UseTLS=True and CACertPath in TRMQConnOptions.
Always declare queues as durable and publish with persistent=True so messages survive a broker restart.
Use SetQoS(ch, 1) before Consume to enable fair dispatch across multiple workers.
Exchange Types
etFanout — broadcast all
etTopic — pattern match
etHeaders — header attrs
Functions
- Connect / ConnectEx
- Disconnect / IsConnected
- OpenChannel / CloseChannel
- SetQoS
- DeclareExchange / DeleteExchange
- DeclareQueue / DeclareQueueEx
- BindQueue / UnbindQueue
- DeleteQueue / PurgeQueue
- GetQueueInfo
- Publish / PublishEx
- Consume / CancelConsumer
- NextMessage
- Ack / Nack / Reject
- TxSelect / TxCommit / TxRollback
- ConfirmSelect / WaitConfirms
- RMQVersion
See Also
- mosquitto (MQTT)
- zmq (broker-less)
- librdkafka (streams)
- redis (pub/sub)