cleveragents-core/.opencode/skills/programming-patterns/references/messaging/message-queue.md

Message Queue Pattern

Problem

When a producer generates work that needs to be processed, calling the consumer directly creates tight coupling and means the producer blocks until processing completes. If the consumer is slow or down, the producer fails too. And if there's more work than one consumer can handle, there's no way to distribute load.

Solution

Place a queue between producers and consumers. Producers enqueue messages and continue immediately. Consumers dequeue messages and process them independently. Key properties:

• Point-to-point — Each message is delivered to exactly one consumer.
• Competing consumers — Multiple consumers can read from the same queue to scale horizontally.
• Durability — Messages persist in the queue until acknowledged.
• Decoupling — Producer and consumer don't need to be alive at the same time.

When to Use

• Distributing work across multiple worker processes
• Smoothing traffic spikes (queue absorbs bursts)
• Background job processing (emails, reports, image processing)
• When producer throughput exceeds consumer throughput
• Cross-service async communication

When to Avoid

• When you need immediate responses (use request-reply)
• When all consumers need to see every message (use pub-sub)
• Simple in-process function calls
• When message ordering is critical and can't be partitioned

Pseudocode

class MessageQueue:
    queue = FIFO buffer

    function enqueue(message):
        queue.push(message)

    function dequeue():
        message = queue.pop()
        return message

    function acknowledge(message):
        mark message as processed

Python Implementation

import threading
import time
import queue
from dataclasses import dataclass, field
from typing import Any


@dataclass
class Message:
    id: int
    body: Any
    attempts: int = 0


class MessageQueue:
    def __init__(self, name: str, max_size: int = 100):
        self.name = name
        self._queue: queue.Queue[Message] = queue.Queue(maxsize=max_size)
        self._next_id = 0
        self._lock = threading.Lock()
        self.enqueued = 0
        self.processed = 0

    def enqueue(self, body: Any) -> Message:
        with self._lock:
            self._next_id += 1
            msg = Message(id=self._next_id, body=body)
            self.enqueued += 1
        self._queue.put(msg)
        return msg

    def dequeue(self, timeout: float = 1.0):
        try:
            msg = self._queue.get(timeout=timeout)
            msg.attempts += 1
            return msg
        except queue.Empty:
            return None

    def acknowledge(self, msg: Message):
        with self._lock:
            self.processed += 1

    @property
    def size(self):
        return self._queue.qsize()


# --- Demo ---
mq = MessageQueue("tasks")
stop_event = threading.Event()
results = []
results_lock = threading.Lock()


def producer(name: str, items: list):
    for item in items:
        msg = mq.enqueue(item)
        print(f"  [{name}] Enqueued msg#{msg.id}: {item}")
        time.sleep(0.02)


def consumer(name: str):
    while not stop_event.is_set():
        msg = mq.dequeue(timeout=0.2)
        if msg is None:
            continue
        # Process
        time.sleep(0.05)
        mq.acknowledge(msg)
        with results_lock:
            results.append((name, msg.id, msg.body))
        print(f"  [{name}] Processed msg#{msg.id}: {msg.body}")


# Start 2 consumers (competing consumers)
consumers = [
    threading.Thread(target=consumer, args=(f"Consumer-{i}",), daemon=True)
    for i in range(1, 3)
]
for c in consumers:
    c.start()

# Produce messages
print("=== Producing messages ===")
producer("Producer", [f"task-{i}" for i in range(1, 7)])

# Wait for processing
time.sleep(1)
stop_event.set()
for c in consumers:
    c.join(timeout=1)

print()
print("=== Results ===")
for worker, msg_id, body in sorted(results, key=lambda x: x[1]):
    print(f"  msg#{msg_id} ({body}) → processed by {worker}")

print()
print(f"Enqueued: {mq.enqueued}, Processed: {mq.processed}, Remaining: {mq.size}")

Output (consumer assignment may vary):

=== Producing messages ===
  [Producer] Enqueued msg#1: task-1
  [Consumer-1] Processed msg#1: task-1
  [Producer] Enqueued msg#2: task-2
  [Consumer-2] Processed msg#2: task-2
  [Producer] Enqueued msg#3: task-3
  [Producer] Enqueued msg#4: task-4
  [Consumer-1] Processed msg#3: task-3
  [Consumer-2] Processed msg#4: task-4
  [Producer] Enqueued msg#5: task-5
  [Producer] Enqueued msg#6: task-6
  [Consumer-1] Processed msg#5: task-5
  [Consumer-2] Processed msg#6: task-6

=== Results ===
  msg#1 (task-1) → processed by Consumer-1
  msg#2 (task-2) → processed by Consumer-2
  msg#3 (task-3) → processed by Consumer-1
  msg#4 (task-4) → processed by Consumer-2
  msg#5 (task-5) → processed by Consumer-1
  msg#6 (task-6) → processed by Consumer-2

Enqueued: 6, Processed: 6, Remaining: 0

Go Implementation

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

type Message struct {
	ID   int
	Body string
}

type MessageQueue struct {
	ch        chan Message
	nextID    int64
	processed int64
}

func NewMessageQueue(size int) *MessageQueue {
	return &MessageQueue{ch: make(chan Message, size)}
}

func (mq *MessageQueue) Enqueue(body string) Message {
	id := int(atomic.AddInt64(&mq.nextID, 1))
	msg := Message{ID: id, Body: body}
	mq.ch <- msg
	return msg
}

func (mq *MessageQueue) Dequeue() (Message, bool) {
	select {
	case msg := <-mq.ch:
		return msg, true
	case <-time.After(200 * time.Millisecond):
		return Message{}, false
	}
}

func (mq *MessageQueue) Ack() {
	atomic.AddInt64(&mq.processed, 1)
}

func main() {
	mq := NewMessageQueue(100)
	var wg sync.WaitGroup
	done := make(chan struct{})

	// Consumers
	for i := 1; i <= 2; i++ {
		wg.Add(1)
		go func(name string) {
			defer wg.Done()
			for {
				select {
				case <-done:
					return
				default:
				}
				msg, ok := mq.Dequeue()
				if !ok {
					select {
					case <-done:
						return
					default:
						continue
					}
				}
				time.Sleep(50 * time.Millisecond)
				mq.Ack()
				fmt.Printf("  [%s] Processed msg#%d: %s\n", name, msg.ID, msg.Body)
			}
		}(fmt.Sprintf("Consumer-%d", i))
	}

	// Producer
	fmt.Println("=== Producing messages ===")
	for i := 1; i <= 6; i++ {
		msg := mq.Enqueue(fmt.Sprintf("task-%d", i))
		fmt.Printf("  [Producer] Enqueued msg#%d: %s\n", msg.ID, msg.Body)
		time.Sleep(20 * time.Millisecond)
	}

	time.Sleep(1 * time.Second)
	close(done)
	wg.Wait()

	fmt.Printf("\nProcessed: %d\n", atomic.LoadInt64(&mq.processed))
}

JavaScript Implementation

class MessageQueue {
  constructor(name) {
    this.name = name;
    this._queue = [];
    this._nextId = 0;
    this.enqueued = 0;
    this.processed = 0;
    this._waiters = [];
  }

  enqueue(body) {
    this._nextId++;
    const msg = { id: this._nextId, body, attempts: 0 };
    this.enqueued++;

    if (this._waiters.length > 0) {
      const resolve = this._waiters.shift();
      msg.attempts++;
      resolve(msg);
    } else {
      this._queue.push(msg);
    }
    return msg;
  }

  async dequeue(timeoutMs = 1000) {
    if (this._queue.length > 0) {
      const msg = this._queue.shift();
      msg.attempts++;
      return msg;
    }
    return new Promise((resolve) => {
      const timer = setTimeout(() => {
        const idx = this._waiters.indexOf(resolve);
        if (idx !== -1) this._waiters.splice(idx, 1);
        resolve(null);
      }, timeoutMs);

      this._waiters.push((msg) => {
        clearTimeout(timer);
        resolve(msg);
      });
    });
  }

  acknowledge() {
    this.processed++;
  }
}

// --- Demo ---
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

(async () => {
  const mq = new MessageQueue("tasks");
  const results = [];

  async function consumer(name, stopSignal) {
    while (!stopSignal.stopped) {
      const msg = await mq.dequeue(200);
      if (!msg) continue;
      await sleep(50);
      mq.acknowledge();
      results.push({ worker: name, id: msg.id, body: msg.body });
      console.log(`  [${name}] Processed msg#${msg.id}: ${msg.body}`);
    }
  }

  const signal = { stopped: false };
  const c1 = consumer("Consumer-1", signal);
  const c2 = consumer("Consumer-2", signal);

  console.log("=== Producing messages ===");
  for (let i = 1; i <= 6; i++) {
    const msg = mq.enqueue(`task-${i}`);
    console.log(`  [Producer] Enqueued msg#${msg.id}: ${msg.body}`);
    await sleep(20);
  }

  await sleep(800);
  signal.stopped = true;
  await Promise.all([c1, c2]);

  console.log(`\nEnqueued: ${mq.enqueued}, Processed: ${mq.processed}`);
})();

Related Patterns

• Publish-Subscribe — publish-subscribe delivers one message to many subscribers; message queue delivers each message to exactly one consumer.
• Producer-Consumer — a message queue is the durable, distributed implementation of the in-process producer-consumer pattern.
• Dead Letter Queue — messages that cannot be processed after N retries are routed to a dead letter queue for inspection.
• Retry — pair message queue consumers with retry logic and backoff to handle transient processing failures.