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:

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);