← Back to glossary

Message Queue Patterns

messaging Intermediate

TL;DR

Core messaging patterns — competing consumers, scatter-gather, request-reply, message filter, and routing slip — solve recurring distributed communication challenges.

Explanation

Key patterns: (1) Competing consumers: multiple consumers on one queue, each processes different messages — horizontal scaling. (2) Scatter-gather: broadcast to multiple services, aggregate responses. (3) Request-reply: producer sends and waits for reply on a correlation ID — makes async look synchronous. (4) Message filter: consumer silectively processes messages matching criteria. (5) Dead letter: failed messages routed to DLQ. (6) Retry with backoff: failed processing retried with increasing delay. (7) Priority queue: high-priority messages processed first. (8) Message expiry: TTL after which message is discarded. These are from Enterprise Integration Patterns (Hohpe & Woolf).

Common Misconception

✗ Queues are just async function calls — queues enable complex patterns (scatter-gather, correlation, routing) far beyond simple async invocation.

Why It Matters

Knowing named patterns enables choosing the right approach and communicating architecture decisions precisely — 'use scatter-gather here' is clearer than describing it from scratch.

Common Mistakes

Request-reply via queue without timeout — hangs forever if reply doesn't come.
Competing consumers on an ordered queue — breaks message ordering.
Priority queue without monitoring — high-priority messages don't help if consumers are busy.

Code Examples

✗ Vulnerable

// Request-reply without timeout/correlation:
$replyQueue = 'reply.' . uniqid();
$producer->send('service.requests', ['reply_to' => $replyQueue, 'data' => $data]);
$reply = $consumer->waitFor($replyQueue); // Hangs if service is down

✓ Fixed

// Request-reply with timeout and correlation ID:
$correlationId = Uuid::v4();
$producer->send('service.requests', [
    'correlation_id' => $correlationId,
    'reply_to' => 'my-service.replies',
    'data' => $data,
]);
$reply = $consumer->waitFor('my-service.replies', $correlationId, timeout: 5);
if (!$reply) throw new ServiceTimeoutException();

References

↗ https://www.enterpriseintegrationpatterns.com/