How I wrote a Go message broker with a throughput of a million messages per second

The reporter describes the process of developing a high-performance Go-based message broker named HermitMQ. The primary objective was to minimize garbage collection and serialization overhead, which can negatively impact performance when handling large message volumes.

To achieve this, the reporter opted for a custom 29-byte binary protocol instead of heavy serialization libraries like JSON. This approach resulted in a highly efficient and lightweight message structure:

type Message struct {

Magic byte

Timestamp uint64

Offset uint64

KeySize uint32

PayloadSize uint32

RecordCount uint32

Key []byte

Payload []byte

}

The header contains a magic number for version checking, timestamps in nanoseconds, offsets for maintaining message order, key and payload sizes, and a record count for batching support.

The message transmission takes place via a direct file-to-socket copy mechanism. When a consumer requests data, the broker looks up the offset in the log file and utilizes the built-in io.CopyN function to transfer data directly from the physical file on disk into the consumer's network socket. This method eliminates the need for reading the entire file into memory, thus minimizing CPU usage and latency spikes.

HermitMQ employs a write-ahead log (WAL) for data storage. The log is divided into segments, each with a .wal file containing raw data and an .idx file that maps offsets to physical bytes on the hard drive. Indices are mapped directly into RAM using mmap for fast searching. A binary search is performed in memory to quickly locate the required byte.

To handle crashes gracefully, the broker includes a torn page recovery mechanism. Upon startup, it scans the WAL files and detects partial messages by comparing the expected message size in the header with the physical size of the file on disk. If a partial message is detected, the broker truncates the corrupted tail at the hardware level using os.Truncate.

Lock sharding is implemented to scale the broker across multiple CPU cores. Consumer group offsets are stored under 256 independent shards, each determined by hashing the key string. This approach evenly distributes incoming requests across available memory, preventing threads from queuing up and degrading performance.