实现一个消息队列

第三十九章：实现一个消息队列

消息队列是现代分布式系统的骨架。Kafka 每天处理数万亿条消息，RabbitMQ 为全球数百万个应用提供路由，Redis Streams 让开发者在一行 XADD 后获得近似消息队列的语义。但当你需要为自己的服务添加异步通信时，真正的问题是：你真的需要 Kafka 吗？

本章将从第一原则出发，探究消息队列的本质：它解决什么问题，为什么这样解决，以及如何用 Go 实现一个具备生产级核心特征的消息队列——从内存队列到文件持久化，从简单生产消费到消费者组和死信队列。

Level 1 · 消息队列存在的理由

三个根本问题：解耦、缓冲、扇出

消息队列解决了分布式系统中最普遍的三类问题：

解耦（Decoupling）

直接调用（无论是 HTTP、gRPC 还是函数调用）创造了时间耦合：调用方必须等待被调用方可用。当用户下单时，如果订单服务必须同步等待：积分服务更新、邮件服务发送确认、风控服务审核、仓储服务分配……任意一个服务变慢或宕机，都会让下单接口超时。

消息队列打破了这种时间耦合。订单服务只需发一条消息到队列就可以立即返回。各下游服务按自己的节奏消费消息。即使邮件服务宕机一小时，消息也安全地积压在队列中，服务恢复后自动消费。

缓冲（Buffering）

流量尖峰是互联网业务的常态。618 活动开始的瞬间，每秒可能有十万个订单。如果所有请求都直接打到数据库，数据库必定崩溃。消息队列是流量整形器：生产者可以瞬时写入大量消息，消费者以数据库能承受的速率（比如每秒 1000 条）稳定消费，自然实现削峰填谷。

扇出（Fan-out）

一条消息需要被多个服务处理，这是事件驱动架构的核心模式。用户注册这个事件，需要：发送欢迎邮件、推送营销短信、初始化用户画像、记录审计日志……如果用发布-订阅模式，注册服务只发一条 user.registered 事件，每个关心这个事件的服务都能独立接收并处理，互不干扰。

什么时候应该自己实现，什么时候用 Kafka/RabbitMQ

用 Kafka 的场景：

• 消息量每天超过百亿条
• 需要跨数据中心的多副本
• 消费者需要回溯任意时间点的历史消息（流式处理）
• 团队有专职的 Kafka 运维人员

用 RabbitMQ 的场景：

• 需要复杂路由（topic exchange、header exchange）
• 消息需要优先级队列
• 需要 AMQP 协议互操作性

自己实现或用简单方案的场景：

• 消息量每天百万级以下
• 服务是单机或少数几台机器
• 团队没有能力运维分布式中间件
• 需要与业务数据库事务绑定（如"写数据库同时写队列"的原子性）

最后一个场景特别重要：事务性消息。如果你的业务逻辑是"用户下单时，订单写入 DB 和消息发送必须同时成功或同时失败"，那么 Kafka 做不到（它不参与你的数据库事务），但你自己基于同一个数据库实现的队列可以做到。

核心抽象：主题、分区、消费者组、偏移量

理解这四个概念是理解任何消息队列的钥匙：

主题（Topic）：消息的逻辑分类，如 order.created、user.registered。生产者向主题发消息，消费者从主题收消息。

分区（Partition）：主题的物理分片。分区是实现并行消费的关键——同一主题的不同分区可以在不同机器上并行消费。分区内的消息有序，分区间无序（这是 Kafka 的核心设计取舍）。

消费者组（Consumer Group）：多个消费者共享一个组 ID，共同消费一个主题。组内每条消息只被一个消费者处理（竞争消费）。不同组之间互相独立，同一条消息会被每个组各消费一次（扇出）。

偏移量（Offset）：消息在分区中的位置序号，单调递增。消费者通过提交偏移量来记录"我消费到哪里了"。崩溃重启后，从上次提交的偏移量继续消费——这是实现 at-least-once 语义的关键。

Level 2 · 核心机制原理

持久化：追加日志（WAL）

消息队列的持久化核心是追加日志（Append-Only Log，即 WAL，Write-Ahead Log）。只追加（append-only）的设计有两个关键优势：

1. 顺序写性能：机械硬盘的顺序写带宽可达 200MB/s，而随机写只有几 MB/s。SSD 的差距虽然较小，但顺序写同样更快且损耗更低。追加日志充分利用了顺序写。

2. 天然不可变性：已写入的消息永远不会被修改，消费者可以安全地并发读取任意位置的数据而不需要锁。

日志文件结构（segment 文件）：
┌─────────────────────────────────────────────────────────┐
│ Msg1 [offset=0, size=128, key=..., value=..., CRC=...]  │
│ Msg2 [offset=1, size=256, key=..., value=..., CRC=...]  │
│ Msg3 [offset=2, size=64,  key=..., value=..., CRC=...]  │
│ ...                                                      │
└─────────────────────────────────────────────────────────┘

索引文件（加速随机读取）：
offset → file_position
0      → 0
1      → 128
2      → 384
...

为什么需要 Segment 文件而不是一个大文件？文件大了之后删除过期消息需要截断（无法高效），Segment 设计让过期删除退化为直接删除旧 Segment 文件——O(1) 操作。

消息确认与 At-Least-Once 语义

"至少一次"（at-least-once）是最常见的消息传递保证：消息至少被处理一次，但可能重复。实现原理：

消费者处理消息的生命周期：
1. 消费者从队列拉取消息（offset=N），在本地内存标记为"处理中"
2. 消费者处理消息（可能成功，可能失败，可能崩溃）
3. 处理成功 → 提交 offset N+1（告诉队列："我已处理到N"）
4. 处理失败/崩溃 → 不提交 offset
5. 下次启动/重试时，从上次提交的 offset 重新消费 → 重复处理消息 N

这是为什么消息处理必须是幂等的（idempotent）：相同消息处理两次，结果与处理一次相同。典型实现：在业务表中用消息 ID 做唯一键，重复插入时忽略（INSERT IGNORE 或 ON CONFLICT DO NOTHING）。

消费者组协调

消费者组的核心难题是：当消费者数量或分区数量发生变化时，如何重新分配分区（Rebalance）？

Kafka 的方案是使用 Group Coordinator（某个 Broker 节点）：组内第一个加入的消费者成为 Leader，负责分区分配计算（用 RangeAssignor 或 RoundRobinAssignor）；然后将分配结果发给 Coordinator；Coordinator 再下发给每个消费者。

简化版方案（适合自实现）：使用 Redis 或数据库做协调状态存储，消费者通过抢锁的方式获得分区所有权，锁有 TTL，消费者崩溃后锁自动释放，其他消费者可以接管。

死信队列

消息处理失败后，不能无限重试（会阻塞整个队列）。死信队列（Dead Letter Queue，DLQ）是解决方案：

正常流程：
Producer → Queue → Consumer（处理成功 → ACK）

失败流程：
Producer → Queue → Consumer（处理失败）
                → 重试 N 次后仍失败
                → 消息移入 DLQ
                → 运维人员手工检查/修复/重放 DLQ 中的消息

DLQ 中的消息携带额外元数据：失败次数、最后一次错误信息、原始队列名、时间戳。这些信息对于排查问题至关重要。

Level 3 · 从零实现消息队列

项目结构

mq/
├── message.go       # 消息结构定义
├── queue.go         # 核心队列接口
├── memory.go        # 内存队列实现
├── persistent.go    # 文件持久化队列
├── producer.go      # 生产者（批量写入）
├── consumer.go      # 消费者（offset 追踪）
├── group.go         # 消费者组
├── dlq.go           # 死信队列
├── server/
│   └── server.go    # HTTP/TCP 管理接口
└── main.go          # 示例

Step 1：消息与队列接口定义

// message.go
package mq

import (
	"encoding/binary"
	"fmt"
	"hash/crc32"
	"time"
)

// Message 是队列中的基本单元。
// 设计原则：消息应该是不可变的，一旦写入就不会被修改。
type Message struct {
	ID        uint64    // 单调递增的全局 ID，即 offset
	Key       []byte    // 可选的分区路由键（相同 key 路由到相同分区）
	Value     []byte    // 消息体，不透明的二进制数据
	Topic     string    // 所属主题
	Partition int       // 所属分区
	Timestamp time.Time // 生产时间
	Headers   map[string]string // 元数据，如追踪 ID、来源服务
	// DLQ 相关字段
	RetryCount  int    // 重试次数
	OrigTopic   string // 原始主题（消息进入 DLQ 时填写）
	LastError   string // 最后一次失败的错误信息
}

// Encode 将消息编码为二进制格式（用于文件持久化）。
// 格式：[4字节总长度][4字节CRC][消息体（JSON或自定义编码）]
// 使用 CRC 校验防止磁盘静默损坏（bit rot）。
func (m *Message) Encode() []byte {
	// 简化：使用 fmt.Sprintf 生成可读格式（生产级应使用 protobuf 或自定义二进制格式）
	body := fmt.Sprintf("%d\x00%s\x00%s\x00%d\x00%d\x00%s",
		m.ID, m.Key, m.Value, m.Timestamp.UnixNano(), m.Partition, m.Topic)
	bodyBytes := []byte(body)

	checksum := crc32.ChecksumIEEE(bodyBytes)
	buf := make([]byte, 8+len(bodyBytes))
	binary.BigEndian.PutUint32(buf[0:4], uint32(len(bodyBytes)))
	binary.BigEndian.PutUint32(buf[4:8], checksum)
	copy(buf[8:], bodyBytes)
	return buf
}

// Queue 是消息队列的核心接口。
// 设计原则：接口应该最小化——只包含不得不放的方法。
type Queue interface {
	// Publish 向指定主题发布消息，返回分配的 offset。
	Publish(topic string, key, value []byte, headers map[string]string) (uint64, error)
	// Consume 从指定主题和分区读取从 offset 开始的最多 maxCount 条消息。
	// 非阻塞：如果没有新消息，返回空切片。
	Consume(topic string, partition int, offset uint64, maxCount int) ([]*Message, error)
	// Commit 提交消费者的 offset（记录"我已消费到这里"）。
	Commit(group, topic string, partition int, offset uint64) error
	// GetOffset 获取消费者组在某分区的当前 offset。
	GetOffset(group, topic string, partition int) (uint64, error)
	// TopicMeta 获取主题的分区信息。
	TopicMeta(topic string) (partitionCount int, err error)
	// Close 关闭队列，释放资源。
	Close() error
}

Step 2：内存队列实现

// memory.go
package mq

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

// MemoryPartition 是单个分区的内存实现。
// 使用切片存储消息，offset 即下标（简化实现，生产级需要 segment 分割和过期清理）。
type MemoryPartition struct {
	mu       sync.RWMutex
	messages []*Message
	// 使用 cond 让消费者等待新消息（避免轮询浪费 CPU）
	cond     *sync.Cond
}

func newMemoryPartition() *MemoryPartition {
	mp := &MemoryPartition{
		messages: make([]*Message, 0, 1024),
	}
	mp.cond = sync.NewCond(&mp.mu)
	return mp
}

// append 追加消息并广播通知所有等待的消费者。
func (mp *MemoryPartition) append(msg *Message) {
	mp.mu.Lock()
	mp.messages = append(mp.messages, msg)
	mp.mu.Unlock()
	mp.cond.Broadcast() // 唤醒所有等待新消息的消费者
}

// read 从指定 offset 读取最多 maxCount 条消息。
// blocking=true 时会等待新消息（long-polling 语义）。
func (mp *MemoryPartition) read(offset uint64, maxCount int, blocking bool) []*Message {
	mp.mu.RLock()

	if blocking {
		// 升级为写锁以使用 cond
		mp.mu.RUnlock()
		mp.mu.Lock()
		for uint64(len(mp.messages)) <= offset {
			mp.cond.Wait() // 释放锁并等待，直到 Broadcast 唤醒
		}
		mp.mu.Unlock()
		mp.mu.RLock()
	}

	defer mp.mu.RUnlock()

	if uint64(len(mp.messages)) <= offset {
		return nil
	}
	end := int(offset) + maxCount
	if end > len(mp.messages) {
		end = len(mp.messages)
	}
	result := make([]*Message, end-int(offset))
	copy(result, mp.messages[offset:end])
	return result
}

// MemoryQueue 是基于内存的消息队列实现。
// 不持久化，重启后数据丢失——适合测试和开发环境。
type MemoryQueue struct {
	mu         sync.RWMutex
	topics     map[string][]*MemoryPartition // topic → partitions
	offsets    map[string]uint64             // "group:topic:partition" → committed offset
	globalID   atomic.Uint64                // 全局单调递增 ID（跨主题唯一）
	partitions int                           // 默认分区数
}

func NewMemoryQueue(defaultPartitions int) *MemoryQueue {
	return &MemoryQueue{
		topics:     make(map[string][]*MemoryPartition),
		offsets:    make(map[string]uint64),
		partitions: defaultPartitions,
	}
}

// ensureTopic 确保主题和其分区已初始化（懒创建）。
func (q *MemoryQueue) ensureTopic(topic string) []*MemoryPartition {
	q.mu.Lock()
	defer q.mu.Unlock()
	if parts, ok := q.topics[topic]; ok {
		return parts
	}
	parts := make([]*MemoryPartition, q.partitions)
	for i := range parts {
		parts[i] = newMemoryPartition()
	}
	q.topics[topic] = parts
	return parts
}

// selectPartition 根据 key 选择分区（key 为空时轮询）。
// 相同 key 的消息总是路由到相同分区，保证有序性。
func (q *MemoryQueue) selectPartition(key []byte, partitionCount int) int {
	if len(key) == 0 {
		// 无 key：轮询（用全局 ID 取模）
		return int(q.globalID.Load() % uint64(partitionCount))
	}
	// 有 key：一致性哈希（简化为 FNV-1a）
	h := uint32(2166136261)
	for _, b := range key {
		h ^= uint32(b)
		h *= 16777619
	}
	return int(h) % partitionCount
}

func (q *MemoryQueue) Publish(topic string, key, value []byte, headers map[string]string) (uint64, error) {
	parts := q.ensureTopic(topic)
	partition := q.selectPartition(key, len(parts))
	id := q.globalID.Add(1) - 1

	msg := &Message{
		ID:        id,
		Key:       key,
		Value:     value,
		Topic:     topic,
		Partition: partition,
		Timestamp: time.Now(),
		Headers:   headers,
	}
	parts[partition].append(msg)
	return id, nil
}

func (q *MemoryQueue) Consume(topic string, partition int, offset uint64, maxCount int) ([]*Message, error) {
	parts := q.ensureTopic(topic)
	if partition >= len(parts) {
		return nil, fmt.Errorf("mq: invalid partition %d for topic %s", partition, topic)
	}
	return parts[partition].read(offset, maxCount, false), nil
}

func (q *MemoryQueue) Commit(group, topic string, partition int, offset uint64) error {
	key := fmt.Sprintf("%s:%s:%d", group, topic, partition)
	q.mu.Lock()
	q.offsets[key] = offset
	q.mu.Unlock()
	return nil
}

func (q *MemoryQueue) GetOffset(group, topic string, partition int) (uint64, error) {
	key := fmt.Sprintf("%s:%s:%d", group, topic, partition)
	q.mu.RLock()
	offset := q.offsets[key]
	q.mu.RUnlock()
	return offset, nil
}

func (q *MemoryQueue) TopicMeta(topic string) (int, error) {
	parts := q.ensureTopic(topic)
	return len(parts), nil
}

func (q *MemoryQueue) Close() error { return nil }

Step 3：文件持久化队列

// persistent.go
package mq

import (
	"bufio"
	"encoding/binary"
	"fmt"
	"hash/crc32"
	"io"
	"os"
	"path/filepath"
	"sync"
)

const (
	segmentMaxSize = 64 * 1024 * 1024 // 64MB per segment
	indexInterval  = 64               // 每 64 条消息建一个索引条目
)

// LogSegment 是单个日志段文件，对应一个 .log 文件（消息数据）和 .idx 文件（稀疏索引）。
type LogSegment struct {
	mu         sync.Mutex
	baseOffset uint64  // 本 segment 第一条消息的 offset
	file       *os.File
	idxFile    *os.File
	writer     *bufio.Writer
	size       int64   // 当前 segment 大小
}

// PersistentPartition 是基于文件的分区实现。
type PersistentPartition struct {
	mu       sync.RWMutex
	dir      string        // 数据目录
	segments []*LogSegment
	nextOffset uint64      // 下一条消息的 offset
}

func newPersistentPartition(dir string) (*PersistentPartition, error) {
	if err := os.MkdirAll(dir, 0755); err != nil {
		return nil, err
	}
	pp := &PersistentPartition{dir: dir}
	// 打开或创建初始 segment
	seg, err := pp.createSegment(0)
	if err != nil { return nil, err }
	pp.segments = []*LogSegment{seg}
	return pp, nil
}

func (pp *PersistentPartition) createSegment(baseOffset uint64) (*LogSegment, error) {
	logPath := filepath.Join(pp.dir, fmt.Sprintf("%020d.log", baseOffset))
	idxPath := filepath.Join(pp.dir, fmt.Sprintf("%020d.idx", baseOffset))

	logFile, err := os.OpenFile(logPath, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0644)
	if err != nil { return nil, err }

	idxFile, err := os.OpenFile(idxPath, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0644)
	if err != nil {
		logFile.Close()
		return nil, err
	}

	stat, _ := logFile.Stat()
	return &LogSegment{
		baseOffset: baseOffset,
		file:       logFile,
		idxFile:    idxFile,
		writer:     bufio.NewWriterSize(logFile, 64*1024),
		size:       stat.Size(),
	}, nil
}

// Write 将消息追加到当前 segment。
// 如果当前 segment 超过大小限制，创建新 segment（滚动）。
func (pp *PersistentPartition) Write(value []byte) (uint64, error) {
	pp.mu.Lock()
	defer pp.mu.Unlock()

	seg := pp.segments[len(pp.segments)-1]
	seg.mu.Lock()
	defer seg.mu.Unlock()

	// 检查是否需要滚动 segment
	if seg.size >= segmentMaxSize {
		seg.writer.Flush()
		newBase := pp.nextOffset
		newSeg, err := pp.createSegment(newBase)
		if err != nil { return 0, err }
		pp.segments = append(pp.segments, newSeg)
		seg = newSeg
	}

	offset := pp.nextOffset

	// 消息格式：[4字节数据长度][4字节CRC32][数据]
	checksum := crc32.ChecksumIEEE(value)
	header := make([]byte, 8)
	binary.BigEndian.PutUint32(header[0:4], uint32(len(value)))
	binary.BigEndian.PutUint32(header[4:8], checksum)

	// 写入数据
	if _, err := seg.writer.Write(header); err != nil { return 0, err }
	if _, err := seg.writer.Write(value); err != nil { return 0, err }

	// 每隔 indexInterval 条消息写一条稀疏索引
	if offset%indexInterval == 0 {
		// 索引条目：[8字节offset][8字节文件位置]
		idxEntry := make([]byte, 16)
		binary.BigEndian.PutUint64(idxEntry[0:8], offset)
		binary.BigEndian.PutUint64(idxEntry[8:16], uint64(seg.size))
		seg.idxFile.Write(idxEntry)
	}

	seg.size += int64(8 + len(value))
	pp.nextOffset++

	// 每 1000 条消息强制 flush（生产级应该根据时间和大小综合决策）
	if offset%1000 == 0 {
		seg.writer.Flush()
	}
	return offset, nil
}

// Read 从指定 offset 读取消息（简化实现：顺序扫描）。
// 生产级实现应通过稀疏索引跳转到近似位置后再顺序扫描。
func (pp *PersistentPartition) Read(startOffset uint64, maxCount int) ([][]byte, error) {
	pp.mu.RLock()
	defer pp.mu.RUnlock()

	var results [][]byte
	currentOffset := uint64(0)

	for _, seg := range pp.segments {
		if seg.baseOffset > startOffset {
			break
		}
		// 强制 flush 确保读到最新数据
		seg.mu.Lock()
		seg.writer.Flush()
		seg.mu.Unlock()

		f, err := os.Open(seg.file.Name())
		if err != nil { return nil, err }
		defer f.Close()

		reader := bufio.NewReader(f)
		offset := seg.baseOffset

		for {
			header := make([]byte, 8)
			if _, err := io.ReadFull(reader, header); err != nil {
				if err == io.EOF { break }
				return nil, err
			}
			size := binary.BigEndian.Uint32(header[0:4])
			expectedCRC := binary.BigEndian.Uint32(header[4:8])

			data := make([]byte, size)
			if _, err := io.ReadFull(reader, data); err != nil { return nil, err }

			// CRC 校验
			actualCRC := crc32.ChecksumIEEE(data)
			if actualCRC != expectedCRC {
				return nil, fmt.Errorf("mq: CRC mismatch at offset %d (data corruption)", offset)
			}

			if offset >= startOffset {
				results = append(results, data)
				if len(results) >= maxCount { return results, nil }
			}
			offset++
			currentOffset = offset
		}
	}
	_ = currentOffset
	return results, nil
}

Step 4：生产者（批量写入）

// producer.go
package mq

import (
	"sync"
	"time"
)

// ProducerConfig 配置生产者的批量行为。
type ProducerConfig struct {
	// BatchSize：积累到这么多条消息后批量发送
	BatchSize int
	// BatchTimeout：即使没有达到 BatchSize，超过这个时间也强制发送
	BatchTimeout time.Duration
	// Async：是否异步发送（true 时 Publish 立即返回，消息在后台发送）
	Async bool
}

type pendingMsg struct {
	topic   string
	key     []byte
	value   []byte
	headers map[string]string
	result  chan<- publishResult
}

type publishResult struct {
	offset uint64
	err    error
}

// Producer 封装了消息的批量发送逻辑。
// 核心思路：收集小消息，积累到一定数量或超时后，一次性写入队列。
// 这减少了锁竞争和 I/O 系统调用次数。
type Producer struct {
	queue   Queue
	config  ProducerConfig
	pending chan *pendingMsg
	done    chan struct{}
	wg      sync.WaitGroup
}

func NewProducer(q Queue, cfg ProducerConfig) *Producer {
	if cfg.BatchSize <= 0 { cfg.BatchSize = 100 }
	if cfg.BatchTimeout <= 0 { cfg.BatchTimeout = 10 * time.Millisecond }

	p := &Producer{
		queue:   q,
		config:  cfg,
		pending: make(chan *pendingMsg, cfg.BatchSize*2),
		done:    make(chan struct{}),
	}
	p.wg.Add(1)
	go p.batchLoop()
	return p
}

// Publish 发布消息。Async=false 时等待消息真正写入队列后返回。
func (p *Producer) Publish(topic string, key, value []byte, headers map[string]string) (uint64, error) {
	resultCh := make(chan publishResult, 1)
	msg := &pendingMsg{
		topic: topic, key: key, value: value,
		headers: headers, result: resultCh,
	}

	select {
	case p.pending <- msg:
	case <-p.done:
		return 0, fmt.Errorf("producer: closed")
	}

	if p.config.Async {
		return 0, nil // 异步模式：立即返回，不等待结果
	}
	r := <-resultCh
	return r.offset, r.err
}

// batchLoop 是生产者的核心 goroutine：收集消息，批量提交。
func (p *Producer) batchLoop() {
	defer p.wg.Done()
	batch := make([]*pendingMsg, 0, p.config.BatchSize)
	ticker := time.NewTicker(p.config.BatchTimeout)
	defer ticker.Stop()

	flush := func() {
		if len(batch) == 0 { return }
		for _, msg := range batch {
			offset, err := p.queue.Publish(msg.topic, msg.key, msg.value, msg.headers)
			if msg.result != nil {
				msg.result <- publishResult{offset: offset, err: err}
			}
		}
		batch = batch[:0]
	}

	for {
		select {
		case msg := <-p.pending:
			batch = append(batch, msg)
			if len(batch) >= p.config.BatchSize {
				flush() // 达到批量大小，立即 flush
			}
		case <-ticker.C:
			flush() // 超时，强制 flush
		case <-p.done:
			// 关闭前处理剩余消息
			for len(p.pending) > 0 {
				batch = append(batch, <-p.pending)
			}
			flush()
			return
		}
	}
}

func (p *Producer) Close() {
	close(p.done)
	p.wg.Wait()
}

Step 5：消费者（offset 追踪）与死信队列

// consumer.go
package mq

import (
	"context"
	"fmt"
	"log"
	"time"
)

// Handler 是消息处理函数。返回 nil 表示处理成功（会 ACK）；返回 error 表示失败（会重试）。
type Handler func(msg *Message) error

// ConsumerConfig 配置消费者行为。
type ConsumerConfig struct {
	Group       string        // 消费者组名
	Topic       string        // 订阅的主题
	Partition   int           // 消费的分区
	MaxRetries  int           // 最大重试次数（超过后进入 DLQ）
	RetryDelay  time.Duration // 重试间隔
	PollInterval time.Duration // 无新消息时的轮询间隔
	BatchSize   int           // 每次拉取的最大消息数
}

// Consumer 从指定主题和分区消费消息，自动管理 offset 和重试逻辑。
type Consumer struct {
	queue  Queue
	dlq    Queue // 死信队列（可以是同一个 Queue 实例的不同 topic）
	config ConsumerConfig
}

func NewConsumer(q Queue, dlq Queue, cfg ConsumerConfig) *Consumer {
	if cfg.MaxRetries <= 0 { cfg.MaxRetries = 3 }
	if cfg.RetryDelay <= 0 { cfg.RetryDelay = 1 * time.Second }
	if cfg.PollInterval <= 0 { cfg.PollInterval = 100 * time.Millisecond }
	if cfg.BatchSize <= 0 { cfg.BatchSize = 10 }
	return &Consumer{queue: q, dlq: dlq, config: cfg}
}

// Run 启动消费循环，直到 ctx 被取消。
// 这是消费者的核心——一个无限循环，拉取消息并调用 handler 处理。
func (c *Consumer) Run(ctx context.Context, handler Handler) error {
	// 从上次提交的 offset 开始消费（实现断点续消费）
	offset, err := c.queue.GetOffset(c.config.Group, c.config.Topic, c.config.Partition)
	if err != nil {
		return fmt.Errorf("consumer: get initial offset: %w", err)
	}

	log.Printf("consumer[%s] starting from offset %d (topic=%s, partition=%d)",
		c.config.Group, offset, c.config.Topic, c.config.Partition)

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}

		// 拉取消息
		msgs, err := c.queue.Consume(c.config.Topic, c.config.Partition, offset, c.config.BatchSize)
		if err != nil {
			log.Printf("consumer: consume error: %v", err)
			time.Sleep(c.config.PollInterval)
			continue
		}

		if len(msgs) == 0 {
			// 没有新消息，等待后继续轮询
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(c.config.PollInterval):
			}
			continue
		}

		// 逐条处理消息（也可以改为批量处理）
		for _, msg := range msgs {
			if err := c.processWithRetry(ctx, msg, handler); err != nil {
				// 超过最大重试次数，进入 DLQ
				c.sendToDLQ(msg, err)
			}
			// 每处理完一条消息就提交 offset
			// 注意：这是 at-least-once 语义——如果进程在提交前崩溃，下次会重复处理这条消息
			// 如果要减少重复，可以批量提交（每 N 条提交一次），但这会增加重复的可能性
			if commitErr := c.queue.Commit(c.config.Group, c.config.Topic, c.config.Partition, msg.ID+1); commitErr != nil {
				log.Printf("consumer: commit offset error: %v", commitErr)
			}
			offset = msg.ID + 1
		}
	}
}

// processWithRetry 处理单条消息，失败时按配置重试。
func (c *Consumer) processWithRetry(ctx context.Context, msg *Message, handler Handler) error {
	var lastErr error
	for attempt := 0; attempt <= c.config.MaxRetries; attempt++ {
		if attempt > 0 {
			// 指数退避（简化版：固定延迟）
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(c.config.RetryDelay * time.Duration(attempt)):
			}
		}
		if err := handler(msg); err == nil {
			return nil // 处理成功
		} else {
			lastErr = err
			msg.RetryCount = attempt + 1
			msg.LastError = err.Error()
			log.Printf("consumer: message %d attempt %d/%d failed: %v",
				msg.ID, attempt+1, c.config.MaxRetries, err)
		}
	}
	return lastErr
}

// sendToDLQ 将处理失败的消息发送到死信队列。
func (c *Consumer) sendToDLQ(msg *Message, err error) {
	if c.dlq == nil { return }
	msg.OrigTopic = msg.Topic
	msg.LastError = err.Error()
	dlqTopic := "dlq." + msg.Topic
	_, dlqErr := c.dlq.Publish(dlqTopic, msg.Key, msg.Value, map[string]string{
		"orig_topic":   msg.OrigTopic,
		"retry_count":  fmt.Sprintf("%d", msg.RetryCount),
		"last_error":   msg.LastError,
		"orig_offset":  fmt.Sprintf("%d", msg.ID),
	})
	if dlqErr != nil {
		log.Printf("consumer: failed to send message %d to DLQ: %v", msg.ID, dlqErr)
	} else {
		log.Printf("consumer: message %d sent to DLQ (%s)", msg.ID, dlqTopic)
	}
}

Step 6：HTTP 管理接口

// server/server.go
package server

import (
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"

	mq "mq"
)

// Server 提供简单的 HTTP 管理接口，用于发布消息、查询 offset、查看主题信息。
type Server struct {
	queue mq.Queue
}

func New(q mq.Queue) *Server {
	return &Server{queue: q}
}

func (s *Server) RegisterRoutes(mux *http.ServeMux) {
	mux.HandleFunc("/publish", s.handlePublish)
	mux.HandleFunc("/consume", s.handleConsume)
	mux.HandleFunc("/offset", s.handleGetOffset)
	mux.HandleFunc("/topic", s.handleTopicMeta)
}

// POST /publish?topic=xxx
// Body: {"key": "...", "value": "..."}
func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	topic := r.URL.Query().Get("topic")
	if topic == "" {
		http.Error(w, "topic required", http.StatusBadRequest)
		return
	}
	var body struct {
		Key   string `json:"key"`
		Value string `json:"value"`
	}
	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	offset, err := s.queue.Publish(topic, []byte(body.Key), []byte(body.Value), nil)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	json.NewEncoder(w).Encode(map[string]interface{}{"offset": offset})
}

// GET /consume?topic=xxx&partition=0&offset=0&count=10&group=mygroup
func (s *Server) handleConsume(w http.ResponseWriter, r *http.Request) {
	topic := r.URL.Query().Get("topic")
	partition, _ := strconv.Atoi(r.URL.Query().Get("partition"))
	offset, _ := strconv.ParseUint(r.URL.Query().Get("offset"), 10, 64)
	count, _ := strconv.Atoi(r.URL.Query().Get("count"))
	if count <= 0 { count = 10 }

	msgs, err := s.queue.Consume(topic, partition, offset, count)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	type msgView struct {
		ID        uint64 `json:"id"`
		Key       string `json:"key"`
		Value     string `json:"value"`
		Partition int    `json:"partition"`
	}
	views := make([]msgView, len(msgs))
	for i, m := range msgs {
		views[i] = msgView{ID: m.ID, Key: string(m.Key), Value: string(m.Value), Partition: m.Partition}
	}
	json.NewEncoder(w).Encode(views)
}

func (s *Server) handleGetOffset(w http.ResponseWriter, r *http.Request) {
	group := r.URL.Query().Get("group")
	topic := r.URL.Query().Get("topic")
	partition, _ := strconv.Atoi(r.URL.Query().Get("partition"))
	offset, err := s.queue.GetOffset(group, topic, partition)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	json.NewEncoder(w).Encode(map[string]interface{}{"offset": offset, "group": group})
}

func (s *Server) handleTopicMeta(w http.ResponseWriter, r *http.Request) {
	topic := r.URL.Query().Get("topic")
	n, err := s.queue.TopicMeta(topic)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	json.NewEncoder(w).Encode(map[string]interface{}{
		"topic": topic, "partitions": n,
	})
}

func (s *Server) ListenAndServe(addr string) error {
	mux := http.NewServeMux()
	s.RegisterRoutes(mux)
	fmt.Printf("MQ server listening on %s\n", addr)
	return http.ListenAndServe(addr, mux)
}

Level 4 · 进阶：日志压缩、精确一次与零拷贝

日志压缩（Log Compaction）

Kafka 的日志压缩是一个精妙的设计：对于同一个 key 的消息，只保留最新的一条。这将消息队列从"事件流"变成了"状态快照"，允许无限期保留数据而不会无限增长。

压缩前（时间轴）：
key=user:1 → {name: Alice}    offset=0
key=user:2 → {name: Bob}      offset=1
key=user:1 → {name: Alice Lee} offset=2
key=user:3 → {name: Charlie}  offset=3
key=user:1 → {DELETE}         offset=4

压缩后（只保留每个 key 的最新值）：
key=user:1 → {DELETE}         (tombstone，保留一段时间后删除)
key=user:2 → {name: Bob}
key=user:3 → {name: Charlie}

实现日志压缩的关键数据结构是offset map：key → 该 key 最新消息的 offset。压缩时，扫描所有消息，只写出每个 key offset map 中记录的那条消息。

// 日志压缩骨架
func compactSegments(segments []*LogSegment) ([]*LogSegment, error) {
	// 第一遍：构建 offset map（每个 key 的最新 offset）
	latestOffset := make(map[string]uint64)
	for _, seg := range segments {
		for _, msg := range seg.readAll() {
			latestOffset[string(msg.Key)] = msg.ID
		}
	}
	// 第二遍：只写出每个 key 在 offset map 中记录的消息
	// （即每个 key 最新的那条）
	newSeg := createNewSegment()
	for _, seg := range segments {
		for _, msg := range seg.readAll() {
			if latestOffset[string(msg.Key)] == msg.ID {
				newSeg.write(msg)
			}
		}
	}
	return []*LogSegment{newSeg}, nil
}

精确一次语义（Exactly-Once）

实现"精确一次"需要两端配合：

幂等生产者：每条消息携带唯一 ID（producerID + sequenceNumber）。服务端检测重复并去重，即使网络重传也不会产生重复消息。

type IdempotentProducer struct {
	producerID uint64
	sequence   atomic.Uint64
}

func (p *IdempotentProducer) Publish(topic string, value []byte) error {
	seq := p.sequence.Add(1)
	headers := map[string]string{
		"producer_id": strconv.FormatUint(p.producerID, 10),
		"sequence":    strconv.FormatUint(seq, 10),
	}
	// 服务端通过 (producerID, sequence) 去重
	return publishWithDedup(topic, value, headers)
}

事务性消费者：消费者在同一个数据库事务中既处理消息（写入业务数据）又提交 offset（写入 offset 表）：

// 事务性消费：处理消息和提交 offset 在同一个 DB 事务中
func processTransactional(db *sql.DB, msg *Message, handler func(*sql.Tx, *Message) error) error {
	tx, err := db.Begin()
	if err != nil { return err }
	defer tx.Rollback()

	// 先检查是否已处理（幂等检查）
	var count int
	tx.QueryRow("SELECT COUNT(*) FROM processed_messages WHERE msg_id=?", msg.ID).Scan(&count)
	if count > 0 { return tx.Commit() } // 已处理，幂等跳过

	// 处理消息
	if err := handler(tx, msg); err != nil { return err }

	// 记录已处理（在同一事务中）
	_, err = tx.Exec("INSERT INTO processed_messages(msg_id, processed_at) VALUES(?,NOW())", msg.ID)
	if err != nil { return err }

	// 提交 offset（在同一事务中）
	_, err = tx.Exec("INSERT INTO mq_offsets(group_id, topic, partition, offset) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE offset=?",
		"mygroup", msg.Topic, msg.Partition, msg.ID+1, msg.ID+1)
	if err != nil { return err }

	return tx.Commit() // 一次提交：业务数据 + 幂等记录 + offset
}

内存映射文件（零拷贝读取）

对于高吞吐量场景，文件 I/O 的数据拷贝路径是：磁盘 → 内核页缓存 → 用户空间 → 内核 socket 缓冲区 → 网络。sendfile 系统调用可以绕过用户空间，直接从页缓存到 socket（零拷贝），是 Kafka 高吞吐的关键之一。

Go 通过 mmap 实现内存映射文件，允许将文件内容直接映射到进程虚拟地址空间：

import "golang.org/x/sys/unix"

// MmapReader 通过 mmap 读取日志文件，避免多次 read() 系统调用。
type MmapReader struct {
	data []byte // mmap 映射的内存区域
	pos  int
}

func NewMmapReader(path string) (*MmapReader, error) {
	f, err := os.Open(path)
	if err != nil { return nil, err }
	defer f.Close()

	stat, _ := f.Stat()
	data, err := unix.Mmap(int(f.Fd()), 0, int(stat.Size()),
		unix.PROT_READ, unix.MAP_SHARED)
	if err != nil { return nil, err }

	return &MmapReader{data: data}, nil
}

func (r *MmapReader) ReadAt(offset int) []byte {
	// 直接通过内存地址访问，无系统调用
	if offset >= len(r.data) { return nil }
	size := int(binary.BigEndian.Uint32(r.data[offset : offset+4]))
	return r.data[offset+8 : offset+8+size]
}

与 Redis Streams 的吞吐量对比

内存队列的吞吐量之所以远高于 Redis Streams，是因为：

1. 没有网络序列化/反序列化开销
2. 没有锁竞争（读写分离）
3. 直接内存访问，CPU 缓存友好

但内存队列的局限是：重启丢数据、单机容量受限。文件队列的吞吐量与 Redis Streams 相当，但没有网络开销，延迟更低。

本章小结

本章从消息队列存在的三个根本原因（解耦、缓冲、扇出）出发，深入探讨了：

• 核心抽象：主题、分区、消费者组、偏移量的设计哲学
• 持久化机制：追加日志（WAL）、Segment 滚动、稀疏索引
• 消息保证：at-least-once 的实现原理，幂等消费的必要性
• 代码实现：从内存队列 → 文件持久化队列 → 批量生产者 → 带重试的消费者 → 死信队列 → HTTP 管理接口

进阶部分讨论了日志压缩的原理、精确一次语义的双端协作实现（幂等生产者 + 事务性消费者），以及内存映射文件（mmap）实现零拷贝读取。

消息队列的本质是时间解耦：让生产者和消费者可以独立运行在不同的时刻。理解了这个本质，就理解了为什么需要持久化、为什么需要 offset、为什么需要消费者组——它们都是在不同维度上保护这种时间解耦的可靠性。