RabbitMQ #

RabbitMQ is the world’s most popular message broker, implementing the AMQP (Advanced Message Queuing Protocol) protocol. Unlike Kafka, which stores all messages like a log, RabbitMQ is a routing-based broker — messages are sent to an exchange, routed based on binding rules, then placed in the appropriate queue. Once a consumer takes and acknowledges a message, it’s deleted from the queue. This model is ideal for task queues, work distribution, and RPC. The dart_amqp package provides a complete AMQP 0-9-1 client for Dart.

RabbitMQ Core Concepts #

flowchart LR
    P["Producer\n(Dart App)"] -->|publish| EX["Exchange\n(direct/topic/fanout)"]
    EX -->|routing key| Q1["Queue: order.new"]
    EX -->|routing key| Q2["Queue: order.payment"]
    EX -->|fanout| Q3["Queue: notification"]
    Q1 -->|deliver| C1["Consumer 1\n(order processor)"]
    Q2 -->|deliver| C2["Consumer 2\n(payment service)"]
    Q3 -->|deliver| C3["Consumer 3\n(email service)"]
    Q3 -->|deliver| C4["Consumer 4\n(sms service)"]

Concept	Description
Exchange	Receives messages from producers, routes them to queues based on rules
Queue	A storage buffer for messages waiting to be consumed
Binding	A rule connecting an exchange to a queue
Routing Key	A label on messages used by the exchange for routing
Acknowledgment	Confirmation from a consumer that a message was processed
Prefetch	The limit of messages delivered to a consumer before an ACK
DLX	Dead Letter Exchange — where rejected or expired messages go

Package Setup #

dart pub add dart_amqp

# pubspec.yaml
dependencies:
  dart_amqp: ^0.2.5

Connecting to RabbitMQ #

import 'package:dart_amqp/dart_amqp.dart';

Future<void> main() async {
  // Basic connection
  final client = Client(
    settings: ConnectionSettings(
      host: 'localhost',
      port: 5672,
      virtualHost: '/',          // default virtual host
      credentials: PlainAuthenticator('guest', 'guest'),
      connectionTimeout: Duration(seconds: 10),
    ),
  );

  // Open a channel — AMQP operations happen through channels
  final channel = await client.channel();
  print('Connected to RabbitMQ');

  await client.close();
}

Connecting with a URL #

// URL format: amqp://username:***@host:port/vhost
final client = Client(
  settings: ConnectionSettings.fromUri(
    Uri.parse('amqp://admin:***@rabbitmq.example.com:5672/production'),
  ),
);

// With SSL/TLS
final clientTLS = Client(
  settings: ConnectionSettings(
    host: 'rabbitmq.example.com',
    port: 5671,  // AMQPS port
    credentials: PlainAuthenticator('admin', 'password'),
    tlsContext: SecurityContext.defaultContext,
  ),
);

Exchange Types #

RabbitMQ supports several exchange types with different routing behaviors:

import 'package:dart_amqp/dart_amqp.dart';

Future<void> deklarasiExchange(Channel channel) async {
  // Direct exchange — routing based on an exact routing key match
  await channel.exchange(
    'order.direct',
    ExchangeType.DIRECT,
    durable: true,  // survives a RabbitMQ restart
  );

  // Topic exchange — routing with wildcards (* = one word, # = many words)
  await channel.exchange(
    'app.events',
    ExchangeType.TOPIC,
    durable: true,
  );

  // Fanout exchange — broadcasts to all bound queues, ignoring routing keys
  await channel.exchange(
    'broadcast',
    ExchangeType.FANOUT,
    durable: true,
  );

  // Headers exchange — routing based on message headers, not routing keys
  await channel.exchange(
    'reports.headers',
    ExchangeType.HEADERS,
    durable: true,
  );
}

Publisher — Sending Messages #

import 'package:dart_amqp/dart_amqp.dart';
import 'dart:convert';

class OrderPublisher {
  late final Client _client;
  late final Channel _channel;

  Future<void> inisialisasi() async {
    _client = Client(
      settings: ConnectionSettings(
        host: 'localhost',
        credentials: PlainAuthenticator('guest', 'guest'),
      ),
    );
    _channel = await _client.channel();

    // Declare the exchange
    await _channel.exchange('order.events', ExchangeType.TOPIC, durable: true);
  }

  Future<void> kirimOrderDibuat({
    required String idOrder,
    required String idPengguna,
    required double total,
  }) async {
    final pesan = {
      'tipe': 'order.dibuat',
      'idOrder': idOrder,
      'idPengguna': idPengguna,
      'total': total,
      'timestamp': DateTime.now().toUtc().toIso8601String(),
    };

    _channel.basicPublish(
      'order.events',          // exchange name
      'order.dibuat',          // routing key
      AmqpMessage.fromBytes(utf8.encode(jsonEncode(pesan)))
        ..properties = (MessageProperties()
          ..contentType = 'application/json'
          ..deliveryMode = DeliveryMode.PERSISTENT  // saved to disk, survives restarts
          ..timestamp = DateTime.now()
          ..messageId = idOrder),
    );

    print('Event order.dibuat sent for $idOrder');
  }

  Future<void> kirimPembayaranDikonfirmasi({
    required String idOrder,
    required String metodePembayaran,
    required double jumlah,
  }) async {
    final pesan = jsonEncode({
      'tipe': 'pembayaran.dikonfirmasi',
      'idOrder': idOrder,
      'metodePembayaran': metodePembayaran,
      'jumlah': jumlah,
    });

    // Topic routing: 'pembayaran.dikonfirmasi' → queues subscribing to this pattern
    _channel.basicPublish(
      'order.events',
      'pembayaran.dikonfirmasi',
      AmqpMessage.fromBytes(utf8.encode(pesan))
        ..properties = (MessageProperties()
          ..contentType = 'application/json'
          ..deliveryMode = DeliveryMode.PERSISTENT),
    );
  }

  Future<void> tutup() => _client.close();
}

Consumer — Receiving Messages #

import 'package:dart_amqp/dart_amqp.dart';
import 'dart:convert';

Future<void> jalankanConsumer() async {
  final client = Client(
    settings: ConnectionSettings(
      host: 'localhost',
      credentials: PlainAuthenticator('guest', 'guest'),
    ),
  );

  final channel = await client.channel();

  // Declare the exchange (idempotent — safe even if it already exists)
  final exchange = await channel.exchange(
    'order.events',
    ExchangeType.TOPIC,
    durable: true,
  );

  // Declare a queue for this consumer
  final queue = await channel.queue(
    'order-processor',    // queue name (empty = generate a random name)
    durable: true,        // survives restarts
    arguments: {
      // Dead Letter Exchange — rejected messages go here
      'x-dead-letter-exchange': 'order.events.dlx',
      // TTL — messages deleted after 24 hours if not consumed
      'x-message-ttl': 86400000,
      // Max queue length
      'x-max-length': 10000,
    },
  );

  // Binding: the queue listens to routing key patterns from the exchange
  await queue.bind(exchange, 'order.*');          // all order events
  await queue.bind(exchange, 'pembayaran.*');     // all payment events

  // Prefetch — the limit of messages delivered before an ACK
  // Prevents one consumer from getting too many messages
  await channel.basicQos(0, 10);  // max 10 messages without ACK

  // Subscribe with manual acknowledgment
  final consumer = await queue.consume(noAck: false);

  print('Consumer running, waiting for messages from queue: ${queue.name}');

  consumer.listen(
    (message) async {
      try {
        final body = utf8.decode(message.payload!);
        final data = jsonDecode(body) as Map<String, dynamic>;

        print('Message received: ${message.routingKey}');
        await prosesPesan(data);

        // ACK — confirm the message was processed successfully → removed from the queue
        message.ack();
      } catch (e) {
        print('Error processing message: $e');

        // NACK — return to the queue for reprocessing
        // requeue: true → put back in the queue
        // requeue: false → goes to the Dead Letter Exchange (if configured)
        message.nack(requeue: false);  // send to the DLX after failing
      }
    },
    onError: (error) {
      print('Error from the channel: $error');
    },
  );

  // Keep the process alive
  await Future.delayed(Duration(hours: 24));
  await client.close();
}

Future<void> prosesPesan(Map<String, dynamic> data) async {
  switch (data['tipe'] as String) {
    case 'order.dibuat':
      print('Processing a new order: ${data['idOrder']}');
      // ... business logic
    case 'pembayaran.dikonfirmasi':
      print('Processing payment confirmation: ${data['idOrder']}');
      // ... business logic
    default:
      print('Unknown message type: ${data['tipe']}');
  }
}

Dead Letter Exchange (DLX) #

The DLX captures messages that failed processing (rejected/expired) for investigation or retries:

import 'package:dart_amqp/dart_amqp.dart';

Future<void> setupDLX(Channel channel) async {
  // 1. Create the DLX exchange
  final dlxExchange = await channel.exchange(
    'order.events.dlx',
    ExchangeType.DIRECT,
    durable: true,
  );

  // 2. Create the DLQ (Dead Letter Queue)
  final dlQueue = await channel.queue(
    'order-processor.dlq',
    durable: true,
    arguments: {
      // Messages in the DLQ are kept for 30 days for investigation
      'x-message-ttl': 2592000000,
    },
  );

  // 3. Bind the DLQ to the DLX
  await dlQueue.bind(dlxExchange, 'order-processor');

  // 4. DLQ consumer for monitoring and manual retries
  final dlqConsumer = await dlQueue.consume(noAck: false);
  dlqConsumer.listen((message) {
    final headers = message.properties?.headers ?? {};
    print('=== Failed Message in DLQ ===');
    print('Reason: ${headers['x-death']?[0]?['reason']}');
    print('Source queue: ${headers['x-death']?[0]?['queue']}');
    print('Death count: ${headers['x-death']?[0]?['count']}');
    print('Payload: ${utf8.decode(message.payload!)}');

    // After investigation, ACK to remove it from the DLQ
    message.ack();
  });
}

Work Queues — Task Distribution #

Work queues distribute heavy tasks across several workers in round-robin fashion:

import 'package:dart_amqp/dart_amqp.dart';
import 'dart:convert';
import 'dart:math';

// Publisher — send tasks
Future<void> kirimTugas(Channel channel, Map<String, dynamic> tugas) async {
  final queue = await channel.queue(
    'tugas-berat',
    durable: true,
    arguments: {'x-dead-letter-exchange': 'dlx'},
  );

  channel.basicPublish(
    '',              // default exchange (direct to queue)
    queue.name,      // routing key = queue name for the default exchange
    AmqpMessage.fromBytes(utf8.encode(jsonEncode(tugas)))
      ..properties = (MessageProperties()
        ..deliveryMode = DeliveryMode.PERSISTENT),
  );
}

// Worker — process tasks
Future<void> jalankanWorker(String workerId) async {
  final client = Client(
    settings: ConnectionSettings(
      host: 'localhost',
      credentials: PlainAuthenticator('guest', 'guest'),
    ),
  );
  final channel = await client.channel();

  final queue = await channel.queue('tugas-berat', durable: true);

  // Prefetch = 1 → each worker only gets 1 task at a time
  // Ensures even distribution based on worker speed
  await channel.basicQos(0, 1);

  final consumer = await queue.consume(noAck: false);
  print('Worker $workerId ready');

  consumer.listen((message) async {
    final tugas = jsonDecode(utf8.decode(message.payload!)) as Map<String, dynamic>;
    print('Worker $workerId working on: ${tugas['nama']}');

    try {
      // Simulate a heavy task
      final durasi = Duration(seconds: Random().nextInt(5) + 1);
      await Future.delayed(durasi);

      print('Worker $workerId finished: ${tugas['nama']} (${durasi.inSeconds}s)');
      message.ack();
    } catch (e) {
      print('Worker $workerId failed: $e');
      message.nack(requeue: false);
    }
  });
}

The RPC Pattern — Request-Reply #

RabbitMQ supports the RPC (Remote Procedure Call) pattern using correlation IDs and reply-to queues:

import 'package:dart_amqp/dart_amqp.dart';
import 'dart:convert';

// RPC Client — sends requests and waits for replies
Future<Map<String, dynamic>> panggilRPC(
  Channel channel,
  String serverQueue,
  Map<String, dynamic> request,
) async {
  // Create a temporary queue to receive the reply
  final replyQueue = await channel.queue('', exclusive: true);
  final correlationId = DateTime.now().microsecondsSinceEpoch.toString();

  final completer = Completer<Map<String, dynamic>>();

  // Listen for the reply
  final consumer = await replyQueue.consume(noAck: true);
  consumer.listen((message) {
    if (message.properties?.correlationId == correlationId) {
      final data = jsonDecode(utf8.decode(message.payload!)) as Map<String, dynamic>;
      completer.complete(data);
    }
  });

  // Send the request
  channel.basicPublish(
    '',
    serverQueue,
    AmqpMessage.fromBytes(utf8.encode(jsonEncode(request)))
      ..properties = (MessageProperties()
        ..replyTo = replyQueue.name
        ..correlationId = correlationId),
  );

  // Wait with a 30-second timeout
  return await completer.future.timeout(
    Duration(seconds: 30),
    onTimeout: () => throw TimeoutException('RPC timeout'),
  );
}

// RPC Server — receives requests and sends replies
Future<void> jalankanRPCServer(Channel channel, String nama) async {
  final queue = await channel.queue('rpc.$nama', durable: true);
  await channel.basicQos(0, 1);

  final consumer = await queue.consume(noAck: false);
  print('RPC Server "$nama" running');

  consumer.listen((message) async {
    try {
      final request = jsonDecode(utf8.decode(message.payload!)) as Map<String, dynamic>;

      // Process the request
      final hasil = await prosesRequest(request);

      // Send the reply to the replyTo queue with the same correlationId
      channel.basicPublish(
        '',
        message.properties!.replyTo!,
        AmqpMessage.fromBytes(utf8.encode(jsonEncode(hasil)))
          ..properties = (MessageProperties()
            ..correlationId = message.properties!.correlationId),
      );

      message.ack();
    } catch (e) {
      // Send an error response
      channel.basicPublish(
        '',
        message.properties!.replyTo!,
        AmqpMessage.fromBytes(utf8.encode(jsonEncode({'error': e.toString()})))
          ..properties = (MessageProperties()
            ..correlationId = message.properties!.correlationId),
      );
      message.nack(requeue: false);
    }
  });
}

RabbitMQ Anti-Patterns #

Not Using Acknowledgment #

// ANTI-PATTERN: auto-acknowledge — messages are lost if a consumer crashes mid-processing
final consumer = await queue.consume(noAck: true); // ✗ messages deleted immediately on delivery

consumer.listen((message) async {
  await prosesBerat(message); // if it crashes here, the message is lost!
  // No ack needed because noAck: true — but the message is already deleted
});

// CORRECT: manual acknowledgment
final consumer = await queue.consume(noAck: false); // ✓

consumer.listen((message) async {
  try {
    await prosesBerat(message);
    message.ack(); // confirm after success
  } catch (e) {
    message.nack(requeue: true); // return to the queue on failure
  }
});

Prefetch Too Large #

// ANTI-PATTERN: a very large prefetch — one consumer dominates
await channel.basicQos(0, 10000); // ✗ one consumer gets 10.000 messages at once
// Other workers get no tasks even though this consumer is slow

// CORRECT: small prefetch (1-10) for fair distribution
await channel.basicQos(0, 1); // ✓ one task per worker before an ACK
// Fast workers get more tasks, slow workers get fewer

Not Declaring Queues as Durable #

// ANTI-PATTERN: a non-durable queue — lost when RabbitMQ restarts
final queue = await channel.queue('penting'); // ✗ durable: false by default

// CORRECT: durable: true and PERSISTENT messages for full durability
final queue = await channel.queue(
  'penting',
  durable: true,  // ✓ the queue survives restarts
);

channel.basicPublish(
  '',
  queue.name,
  AmqpMessage.fromBytes(data)
    ..properties = (MessageProperties()
      ..deliveryMode = DeliveryMode.PERSISTENT),  // ✓ messages saved to disk
);

Summary #

Exchange types determine the routing: direct (exact key), topic (wildcards * and #), fanout (broadcast to all bound queues), headers (based on headers).
Idempotent declarations — always declare exchanges and queues in both producers and consumers because startup order is unpredictable.
Manual acknowledgment (noAck: false) is the standard for critical tasks — ack() after success, nack(requeue: false) to send to the DLX on failure.
Prefetch 1 (basicQos(0, 1)) for fair work queues — fast workers get more tasks than slow ones, not blind round-robin.
Durable queues + PERSISTENT messages are needed together for restart-surviving messages — a durable queue with non-persistent messages loses them on restart.
Dead Letter Exchanges (DLX) for capturing failed messages — configure via the x-dead-letter-exchange queue argument and create a DLQ consumer for monitoring.
Topic exchanges are the most flexible — order.* matches order.dibuat, order.dikirim but not order.pembayaran.sukses. order.# matches everything starting with order..
RPC with correlationIds — use a temporary reply queue (exclusive: true) and a unique correlationId to match requests with the right replies.
Prefetch under 10 for consumers doing heavy I/O — consumers won’t be overloaded when operations are slow.
Virtual hosts separate environments within one RabbitMQ instance — use /production, /staging, /development for environment isolation.

← Previous: Kafka Next: Amazon SQS →