Designing an Industrial‑Grade Message Queue for Tens of Millions of Orders

1. Problem Background

In high‑concurrency e‑commerce systems a message queue must do more than simple async buffering: it decouples services, smooths traffic spikes, and guarantees reliable delivery even when nodes crash, networks jitter, or processes restart.

During a mid‑year promotion the platform faced daily order volume of ~30 M, peak write throughput >60 k TPS, and a legacy MySQL task‑table + scan‑retry solution that broke under load, showing lock contention, high latency (P99 from 200 ms to >15 s), lost acknowledgments, out‑of‑order processing, and ineffective scaling.

2. Business Scenario – Order Event Bus

Events such as OrderCreated, InventoryReserved, PaymentSucceeded, OrderDelivered, OrderCancelled, and OrderRefunded are emitted and consumed by inventory, marketing, payment, settlement, notification, and risk‑control subsystems.

Strict ordering per orderId is required.

At‑least‑once delivery is the baseline.

Some flows need transactional consistency between DB write and event emission.

Peak traffic can spike several‑fold within seconds.

Consumer processing times vary widely; the broker must not be throttled by the slowest consumer.

Full observability of write, replication, and consumption state is mandatory.

3. Core Capabilities of an Industrial‑Grade MQ

3.1 Production Semantics

What does a successful produce mean? Memory write, page‑cache flush, disk flush, or replica acknowledgment.

How to avoid duplicate writes after retries?

3.2 Storage Semantics

Message layout, sequential append, random access, batch reads, retention, and cleaning policies.

3.3 Consumption Semantics

Push vs pull, offset management, at‑least‑once, retry, dead‑letter handling, ordered consumption pause.

3.4 High‑Availability Semantics

Leader failure handling, HW (high watermark) vs LEO (log end offset), safe leader election.

3.5 Engineering Governance

Rate limiting, back‑pressure, configuration, rolling upgrades, capacity planning, fault‑injection drills.

4. Design Goals & Trade‑offs

Single‑broker node should sustain 100 k TPS writes.

P99 end‑to‑end latency < 50 ms under normal load.

Strict per‑partition ordering.

At‑least‑once delivery with idempotent consumption.

Transactional messages coordinated with local DB transactions.

Replica replication and automatic leader election.

Observability, rate limiting, retry, and gray‑release capabilities.

Trade‑offs:

Consistency > absolute throughput (ACK=ALL preferred).

Partition‑level ordering > global ordering (shardingKey = orderId).

Pull model for consumers to enable back‑pressure.

Weakly consistent routing metadata, strongly consistent partition logs.

Default at‑least‑once; exactly‑once requires external state binding.

5. Overall Architecture

Control Plane : NameServer stores topic/partition metadata, handles heartbeats, and publishes routing.

Data Plane : Broker stores logs, performs replication, and serves pull requests.

Client Layer : Producer SDK and Consumer SDK.

Governance Layer : Monitoring, alerting, rate limiting, config center, operational tools.

5.1 Storage Layout – SegmentLog

Each partition is a logical log split into multiple .log, .idx, and .timeidx files:

/data/hermes/topic-order/partition-03/00000000000000000000.log
 00000000000000000000.idx
 00000000000000000000.timeidx
 00000000000001048576.log
 00000000000001048576.idx
 00000000000001048576.timeidx

.log

: sequential message bytes. .idx: sparse offset → file position index. .timeidx: time‑based index for fast range scans.

5.2 Message Record Format

+-------------+-------------+-------------+-------------+
| totalSize   | magic       | crc32       | attributes  |
+-------------+-------------+-------------+-------------+
| topicId     | partitionId | queueOffset | bornTime    |
+-------------+-------------+-------------+-------------+
| storeTime   | producerId  | msgId       | keyLength   |
+-------------+-------------+-------------+-------------+
| bodyLength  | headersLen  | key/body/headers bytes ... |
+----------------------------------------------------------+

Key fields: queueOffset: logical offset within the partition (consumer cursor). crc32: integrity check for disk or replication corruption. attributes: flags for compression, transactional half‑message, delayed delivery, etc. msgId: globally unique identifier for tracing and idempotence.

5.3 Sparse Index Rationale

Full indexing would bloat files and increase write amplification. By recording an index entry every N messages or bytes, the system can binary‑search to the nearest entry and then scan a small range, achieving small index size, simple writes, batch‑read friendliness, and fast recovery.

6. Write Path – From Producer to Replica Confirmation

6.1 End‑to‑End Flow

Producer sends PUT(topic, partitionKey, message) → NameServer routes to target partition → Broker leader appends to active Segment → AppendResult returns physical offset and queue offset → Replication to followers (pull model) → High Watermark (HW) advances → ACK sent to producer based on configured mode (0, 1, ALL).

6.2 LEO vs HW

LEO

(Log End Offset): latest offset written on a replica. HW (High Watermark): highest offset replicated to a quorum and visible to consumers.

Producer ACK does not guarantee all replicas have persisted; consumers can only read up to HW.

6.3 ACK Strategies

ACK=0

: fire‑and‑forget, highest throughput, lowest reliability. ACK=1: leader‑local write confirmed, moderate reliability. ACK=ALL: majority replica confirmation, highest reliability (recommended for order‑critical paths).

7. Network Layer – Netty + Reactor + Long‑Lived Connections

Characteristics:

Many concurrent connections (producer & consumer long‑lived sockets).

Small request packets at high frequency, occasional large batch payloads.

Need tight memory, thread‑switch, and back‑pressure control.

Implementation uses Netty’s BossGroup for connection acceptance and WorkerGroup for I/O, separating business thread pools from I/O to avoid blocking.

7.2 Why Not One Thread Per Connection

Blocking I/O would cause high thread‑switch cost, large stack memory, and difficulty handling slow consumers. NIO + event‑driven model solves these issues.

7.3 Multi‑Level Back‑Pressure

Connection‑level request‑count limit.

Topic‑level byte‑per‑second quota.

Broker‑level dynamic throttling based on memory water‑mark, flush latency, and replication lag.

When limits are exceeded the broker returns explicit error codes (THROTTLED, LEADER_NOT_AVAILABLE, REPLICA_LAGGING, DISK_FULL) and the client performs exponential back‑off.

8. Partition Model & Strict Ordering

Global ordering is infeasible; instead, partition‑level ordering is enforced:

Messages with the same orderId are hashed to the same partition: partition = hash(orderId) % partitionCount.

Within a partition, writes are serialized and consumers process a single thread per partition.

8.1 Ordered Consumption Steps

Pull a batch of messages from a partition.

Process them sequentially in a single thread.

On success, commit the next offset.

On failure, pause the partition, trigger retry or dead‑letter handling, and stop further consumption.

9. Consumption Model – Pull, Commit, Retry, Dead‑Letter

9.1 Pull Model Rationale

Pull lets consumers control their own pace, simplifies back‑pressure, and makes slow consumers harmless to the broker.

9.2 Offset Commit Strategies

Automatic periodic commit (simple but may lose in‑flight messages).

Manual commit after successful business processing (recommended for core order flow).

9.3 Retry & Dead‑Letter Topics

Failed messages are moved to order-events.retry.1m, order-events.retry.10m, and finally to order-events.dlq. This provides controlled back‑off, auditability, and prevents a single bad message from blocking the partition.

10. Transactional Messaging

Problem: a DB insert succeeds but the MQ publish fails, or vice‑versa, causing inconsistency.

Solution: Half‑Message (prepare) is written to the log but hidden from consumers. After the local DB transaction commits, the producer sends a commit command; otherwise a rollback is issued. The broker periodically checks the transaction state (via TransactionCoordinator) to resolve uncertain half‑messages.

10.1 Transaction Coordinator Skeleton

public final class TransactionCoordinator {
    private final Map<String, TransactionState> states = new ConcurrentHashMap<>();
    public void markPrepared(String txId) { states.put(txId, TransactionState.PREPARED); }
    public void commit(String txId) { states.put(txId, TransactionState.COMMITTED); }
    public void rollback(String txId) { states.put(txId, TransactionState.ROLLED_BACK); }
    public boolean needCheck(String txId, Duration timeout) {
        TransactionState state = states.get(txId);
        return state == TransactionState.PREPARED;
    }
    public enum TransactionState { PREPARED, COMMITTED, ROLLED_BACK }
}

In production the state must be persisted (e.g., a dedicated transaction log) so that broker restarts can recover pending transactions.

11. High‑Availability Design

11.1 Replica Group

1 Leader + 2+ Followers per partition.

Leader handles all writes and serves reads.

Followers pull logs from the leader.

11.2 Replication Mode – Follower Pull

Pull simplifies flow control; followers fetch at their own pace, making network recovery easier and reducing leader state management.

11.3 Leader Election Rules

Only in‑sync replicas (ISR) are eligible.

New leader must have HW ≥ old leader’s HW.

Log entries beyond HW that lack quorum may be truncated.

11.4 Failure Recovery Workflow

When a leader heartbeat is missed, the control plane selects a new leader from ISR, publishes new metadata, and clients reconnect. The old leader, upon recovery, compares HW/LEO, truncates any orphan logs, and rejoins the replica group.

12. Core Data Structures & Module Boundaries

hermesmq/
  hermes-common/
  hermes-remoting/
  hermes-store/
  hermes-broker/
  hermes-client/
  hermes-nameserver/
  hermes-observability/

Key classes: TopicPartition: identifies a topic and partition. MessageRecord: immutable message payload. AppendResult: result of a log append (offsets, bytes written, flush status). Segment: single log file with synchronized append. PartitionLog: manages active segment and next offset. ReplicaManager: handles replication and HW advancement. PutMessageService: validates flow control, routes to PartitionLog, and coordinates ACK. OrderedConsumeWorker: guarantees single‑threaded ordered processing and integrates retry/DLQ logic. TransactionCoordinator: tracks half‑message state.

13. Production‑Level Core Code Samples

13.1 Message Entity & Append Result

package com.hermesmq.store.model;

public final class MessageRecord {
    private final String topic;
    private final int partition;
    private final String messageKey;
    private final String messageId;
    private final byte[] body;
    private final Map<String, String> headers;
    private final long bornTimestamp;
    private final long producerSequence;
    // constructor, getters, and helper methods omitted for brevity
}

public record AppendResult(long queueOffset, long physicalOffset, int wroteBytes, boolean flushed, long highWatermark) {}

13.2 Segment Append (synchronized for order)

public final class Segment implements AutoCloseable {
    private final long baseOffset;
    private final FileChannel logChannel;
    private final AtomicLong writePosition = new AtomicLong();

    public synchronized AppendResult append(long queueOffset, ByteBuffer encoded, boolean forceFlush, long highWatermark) throws IOException {
        long physicalOffset = writePosition.get();
        int wroteBytes = encoded.remaining();
        while (encoded.hasRemaining()) {
            logChannel.write(encoded, physicalOffset + (wroteBytes - encoded.remaining()));
        }
        writePosition.addAndGet(wroteBytes);
        if (forceFlush) {
            logChannel.force(false);
        }
        return new AppendResult(queueOffset, physicalOffset, wroteBytes, forceFlush, highWatermark);
    }
    // close, getters omitted
}

13.3 PartitionLog – Offset Allocation & Encoding

public final class PartitionLog {
    private final TopicPartition topicPartition;
    private final Segment activeSegment;
    private final AtomicLong nextOffset;

    public AppendResult append(MessageRecord record, boolean forceFlush, long highWatermark) throws IOException {
        long queueOffset = nextOffset.getAndIncrement();
        ByteBuffer encoded = MessageCodec.encode(record, queueOffset, System.currentTimeMillis());
        return activeSegment.append(queueOffset, encoded, forceFlush, highWatermark);
    }
    // getters omitted
}

13.4 PutMessageService – Write + Replication

public final class PutMessageService {
    private final ConcurrentHashMap<TopicPartition, PartitionLog> partitionLogs;
    private final ReplicaManager replicaManager;
    private final BrokerFlowController flowController;

    public AppendResult put(MessageRecord record, AckMode ackMode) throws IOException {
        flowController.checkWritable(record.topic(), record.body().length);
        TopicPartition tp = new TopicPartition(record.topic(), record.partition());
        PartitionLog pl = partitionLogs.get(tp);
        if (pl == null) throw new IllegalStateException("partition not found: " + tp);
        AppendResult local = pl.append(record, ackMode == AckMode.ALL, replicaManager.highWatermark(tp));
        long confirmedHw = replicaManager.replicateAndWait(tp, local, ackMode);
        return new AppendResult(local.queueOffset(), local.physicalOffset(), local.wroteBytes(), local.flushed(), confirmedHw);
    }
}

13.5 OrderedConsumeWorker – Idempotent Processing

public final class OrderedConsumeWorker {
    private final OffsetStore offsetStore;
    private final MessageHandler messageHandler;
    private final RetryPublisher retryPublisher;
    private final DeadLetterPublisher deadLetterPublisher;

    public void consume(TopicPartition partition, List<ConsumerMessage> messages) {
        for (ConsumerMessage msg : messages) {
            try {
                messageHandler.handle(msg);
                offsetStore.commit(partition, msg.queueOffset() + 1);
            } catch (Exception ex) {
                if (msg.retryTimes() >= 3) {
                    deadLetterPublisher.publish(msg, ex.getMessage());
                } else {
                    retryPublisher.publish(msg.nextRetry());
                }
                break; // stop consuming this batch on failure
            }
        }
    }
}

14. API Design – Produce, Pull, Commit, Transaction Check

PUT /api/v1/messages

– JSON body with topic, shardingKey, messageId, headers, base64 body, and ackMode (ALL/1/0). Returns partition, queueOffset, messageId, and highWatermark.

GET /api/v1/messages/fetch?topic=&partition=&offset=&maxMessages=

– long‑poll pull. POST /api/v1/offsets/commit – consumer group offset commit. POST /api/v1/transactions/check – broker‑side transaction state query (COMMIT/ROLLBACK).

15. High‑Concurrency Optimizations

Batching (batch.size, linger.ms, compression.type) to amortize RPC and syscalls.

Partition‑level parallelism: different partitions write concurrently, each partition serialized.

Separate thread pools for I/O, write scheduling, replication, long‑poll notifications, and background cleanup.

Hot‑data on local SSD, cold segments archived to object storage for long‑term retention.

16. Idempotence & Exactly‑Once Guarantees

Because at‑least‑once is guaranteed, consumers must be idempotent:

Database unique keys (e.g., (order_id, event_type)).

Status‑machine checks (skip already‑PAID events).

Deduplication tables keyed by messageId or business key.

Outbox pattern is recommended when native transaction messaging is not feasible.

17. Exception Handling & Compensation

Producer side: business idempotent keys, request IDs, retry only on transient errors.

Consumer side: distinguish retryable vs non‑retryable exceptions, route to retry topics, dead‑letter after max attempts, and provide manual replay with audit metadata.

18. Observability & Operations

Broker metrics: write/read TPS, latency percentiles, flush time, replica lag, HW‑LEO gap, disk usage, page‑cache hit ratio, network queue depth.

Consumer metrics: group lag, consume TPS, retry/DLQ counts, per‑message processing latency, failure reason distribution.

All messages carry

traceId, messageId, topic, partition, queueOffset, consumerGroup

for end‑to‑end tracing.

19. Security & Governance

AK/SK or mTLS authentication for producers/consumers.

Topic‑level ACLs.

Message size limits, header/body validation, compression‑bomb protection.

Multi‑tenant quotas on topics, write rate, and connections.

20. Configuration Management & Gray‑Release

Key YAML snippet (excerpt) shows broker ack mode, max message size, long‑poll timeout, disk alarm threshold, replica ISR count, batch sizes, linger, and request timeout. All parameters are externalized.

Upgrade path: upgrade NameServer first, then roll followers, finally switch and upgrade leader, validating replication lag, write latency, and consumer lag at each step. Client SDKs are rolled out gradually (few producers → few consumers → full rollout).

21. Performance Pitfalls & Common Issues

Page‑cache jitter caused by massive historical replay or insufficient memory.

Disk‑full cascades: increased flush latency, replication slowdown, producer timeouts, traffic avalanche.

Over‑eager synchronous flush (ACK=ALL on every message) kills throughput; use per‑topic policies.

22. Testing & Validation

Benchmark: per‑partition write TPS, multi‑partition concurrency, write P99 latency, consumer lag.

Fault injection: leader kill, follower network partition, disk full, NameServer outage, consumer thread pool saturation. Verify no message loss, ordering preservation, fast leader election, lag recovery.

Transactional tests: half‑message persisted then app crash, commit missing, broker check timeout, duplicate commit/rollback.

23. Evolution Roadmap

Stage 1 – Single‑node, single replica (validate log format, pull model, basic offset).

Stage 2 – Leader‑follower replication, HW/LEO, automatic failover.

Stage 3 – Transactional half‑messages, retry/DLQ, full monitoring.

Stage 4 – Cloud‑native deployment, container orchestration, auto‑scaling, hot‑cold tiered storage, multi‑tenant isolation.

24. Applicability & Boundaries

Suitable for order, payment, inventory, and other transaction‑critical event streams that need strict per‑order ordering, transactional guarantees, and fine‑grained governance.

Not suitable for pure log‑collection at massive scale, teams lacking infrastructure expertise, or scenarios demanding global multi‑region active‑active replication without dedicated operational investment.

25. Conclusion

Building an industrial‑grade MQ is not merely about sending messages; it is about constructing a trustworthy data pipeline that guarantees write semantics, durable sequential storage, consistent replication, ordered consumption, transactional safety, and full observability. When these capabilities are designed as a closed loop, the MQ becomes the backbone of a reliable order processing system.