Message Queue Patterns — RabbitMQ, Kafka, Pub/Sub, Work Queues

// work-queue.js
// Work queue with competing consumers using RabbitMQ (amqplib)

const amqp = require('amqplib');

async function setupWorkQueue() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Durable queue survives broker restarts
  const queue = 'image_tasks';
  await channel.assertQueue(queue, { durable: true });

  // Fair dispatch: don't send more than 1 message at a time per consumer
  channel.prefetch(1);

  console.log('[Work Queue] Waiting for tasks. Press Ctrl+C to exit.');

  channel.consume(queue, (msg) => {
    if (msg === null) return;

    const task = JSON.parse(msg.content.toString());
    console.log(`[Consumer] Processing task: ${task.id} (${task.type})`);

    // Simulate processing time based on task complexity
    const processingTime = task.complexity * 1000;
    setTimeout(() => {
      console.log(`[Consumer] Completed task: ${task.id} in ${processingTime}ms`);
      channel.ack(msg);  // Acknowledge after successful processing
    }, processingTime);
  }, { noAck: false });
}

setupWorkQueue().catch(console.error);

// ── Publisher ──
async function publishTasks() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  const queue = 'image_tasks';
  await channel.assertQueue(queue, { durable: true });

  const tasks = [
    { id: 'T1', type: 'thumbnail', complexity: 2 },
    { id: 'T2', type: 'watermark', complexity: 5 },
    { id: 'T3', type: 'compress', complexity: 1 },
    { id: 'T4', type: 'thumbnail', complexity: 3 },
    { id: 'T5', type: 'exif_clean', complexity: 4 },
  ];

  for (const task of tasks) {
    channel.sendToQueue(queue, Buffer.from(JSON.stringify(task)), {
      persistent: true,  // Message survives broker restart
    });
    console.log(`[Publisher] Sent task: ${task.id}`);
  }

  await channel.close();
  await connection.close();
}

publishTasks().catch(console.error);

# kafka_streaming.py
# Kafka producer and consumer for event streaming

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
import json
import time

# ── Kafka Producer ──

class OrderEventProducer:
    """Producer for order events in Kafka."""

    def __init__(self, bootstrap_servers='localhost:9092'):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            acks='all',  # Wait for all replicas to acknowledge
            retries=3,
            linger_ms=10,  # Batch small messages for 10ms
        )
        self.topic = 'order_events'

    def publish_order_event(self, event_type, data):
        """Publish an order event to Kafka."""
        event = {
            'event_type': event_type,
            'timestamp': time.time(),
            'data': data,
        }

        # Use order_id as partition key for ordering guarantees
        order_id = data.get('order_id', 'unknown')
        future = self.producer.send(
            self.topic,
            key=order_id.encode('utf-8'),
            value=event
        )

        try:
            record_metadata = future.get(timeout=10)
            print(f"[Kafka] Published {event_type} for {order_id} "
                  f"-> partition {record_metadata.partition} "
                  f"offset {record_metadata.offset}")
        except KafkaError as e:
            print(f"[Kafka] Failed to publish: {e}")

    def close(self):
        self.producer.flush()
        self.producer.close()

producer = OrderEventProducer()
producer.publish_order_event('order_created', {
    'order_id': 'ORD-789',
    'user_id': 42,
    'amount': 199.99,
    'items': 3,
})
producer.publish_order_event('payment_processed', {
    'order_id': 'ORD-789',
    'status': 'success',
    'transaction_id': 'txn_abc123',
})
producer.close()

# kafka_consumer.py
# Kafka consumer with consumer groups

from kafka import KafkaConsumer, TopicPartition
import json

class OrderEventConsumer:
    """Consumer for order events with offset management."""

    def __init__(self, group_id='order-processor', bootstrap_servers='localhost:9092'):
        self.consumer = KafkaConsumer(
            'order_events',
            bootstrap_servers=bootstrap_servers,
            group_id=group_id,
            auto_offset_reset='earliest',
            enable_auto_commit=True,
            auto_commit_interval_ms=5000,
            value_deserializer=lambda v: json.loads(v.decode('utf-8')),
        )
        print(f"[Kafka Consumer] Started group={group_id}")

    def process_events(self):
        """Consume and process order events."""
        for message in self.consumer:
            event = message.value
            print(f"[Consumer] Event: {event['event_type']} "
                  f"(partition={message.partition}, "
                  f"offset={message.offset})")

            # Process based on event type
            if event['event_type'] == 'order_created':
                self.handle_order_created(event['data'])
            elif event['event_type'] == 'payment_processed':
                self.handle_payment_processed(event['data'])

    def handle_order_created(self, data):
        print(f"  -> Order {data['order_id']}: ${data['amount']}, {data['items']} items")

    def handle_payment_processed(self, data):
        print(f"  -> Payment {data['transaction_id']}: {data['status']}")

if __name__ == '__main__':
    consumer = OrderEventConsumer()
    consumer.process_events()

# topic_routing.py
# RabbitMQ topic exchange for selective message routing

import pika
import JSON

class TopicRouter:
    """Route messages based on routing key patterns."""

    def __init__(self):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters('localhost', 5672)
        )
        self.channel = self.connection.channel()
        self.channel.exchange_declare(
            exchange='system_events',
            exchange_type='topic',
            durable=True
        )

    def publish(self, routing_key, data):
        """Publish message with a routing key (e.g., 'error.auth.login')."""
        message = {
            'severity': routing_key.split('.')[0],
            'source': '.'.join(routing_key.split('.')[1:]),
            'data': data,
            'timestamp': __import__('time').time(),
        }
        self.channel.basic_publish(
            exchange='system_events',
            routing_key=routing_key,
            body=JSON.dumps(message),
            properties=pika.BasicProperties(delivery_mode=2)
        )
        print(f"[Router] Published: {routing_key}")

    def setup_consumer(self, Queue_name, binding_key):
        """
        Bind a Queue with a pattern.
        * matches one word, # matches zero or more words.
        Example: 'error.#' matches 'error.auth.login'
        """
        result = self.channel.Queue_declare(Queue=Queue_name, durable=True)
        self.channel.Queue_bind(
            exchange='system_events',
            Queue=Queue_name,
            routing_key=binding_key
        )
        print(f"[Router] Queue '{Queue_name}' bound with '{binding_key}'")

        def callback(ch, method, properties, body):
            event = JSON.loads(body)
            print(f"[{Queue_name}] Received: {method.routing_key} — {event['data']}")
            ch.basic_ack(delivery_tag=method.delivery_tag)

        self.channel.basic_consume(
            Queue=Queue_name,
            on_message_callback=callback
        )
        return Queue_name

router = TopicRouter()

# Set up consumers with different patterns
router.setup_consumer('all_errors', 'error.#')
router.setup_consumer('auth_events', '#.auth.#')
router.setup_consumer('critical_alerts', 'critical.*')

# Publish events
router.publish('error.auth.login', {'user': 'alice', 'reason': 'invalid_password'})
router.publish('error.db.timeout', {'query': 'SELECT * FROM orders'})
router.publish('info.auth.logout', {'user': 'bob'})
router.publish('critical.disk', {'usage': 97, 'device': '/dev/sda1'})

// event-bus-amqp.js
// RabbitMQ-based event bus for Microservices

const amqp = require('amqplib');

class EventBus {
  constructor(URL = 'amqp://localhost') {
    this.URL = URL;
    this.connection = null;
    this.channel = null;
    this.exchange = 'app_events';
  }

  async connect() {
    this.connection = await amqp.connect(this.URL);
    this.channel = await this.connection.createChannel();
    await this.channel.assertExchange(this.exchange, 'topic', { durable: true });
    console.log('[EventBus] Connected to RabbitMQ');
    return this;
  }

  async publish(eventType, data, routingKey) {
    if (!this.channel) await this.connect();
    const message = {
      eventType,
      timestamp: new Date().toISOString(),
      data,
    };
    const key = routingKey || eventType.replace(/\./g, '.');
    this.channel.publish(
      this.exchange,
      key,
      Buffer.from(JSON.stringify(message)),
      { persistent: true }
    );
    console.log(`[EventBus] Published: ${eventType} (${key})`);
  }

  async subscribe(patterns, handler, queueName = '') {
    if (!this.channel) await this.connect();
    const q = await this.channel.assertQueue(queueName || '', {
      exclusive: !queueName,
      durable: !!queueName,
    });

    for (const pattern of patterns) {
      await this.channel.bindQueue(q.Queue, this.exchange, pattern);
      console.log(`[EventBus] Bound ${q.Queue} <- ${pattern}`);
    }

    await this.channel.consume(q.Queue, (msg) => {
      if (msg === null) return;
      const event = JSON.parse(msg.content.toString());
      handler(event, msg);
      this.channel.ack(msg);
    });

    return q.Queue;
  }

  async close() {
    await this.channel?.close();
    await this.connection?.close();
  }
}

async function demo() {
  const bus = new EventBus();
  await bus.connect();

  // Subscribe to all user events
  await bus.subscribe(['user.#'], (event) => {
    console.log(`[User Service] ${event.eventType}: ${event.data.email}`);
  });

  // Subscribe to payment events only
  await bus.subscribe(['payment.completed'], (event) => {
    console.log(`[Payment Service] Completed: ${event.data.amount}`);
  });

  // Publish events
  await bus.publish('user.registered', { email: 'alice@example.com', plan: 'premium' });
  await bus.publish('payment.completed', { userId: 1, amount: 29.99 });
  await bus.publish('user.upgraded', { email: 'alice@example.com', newPlan: 'enterprise' });

  setTimeout(() => bus.close(), 1000);
}

demo().catch(console.error);