ozine's book library

主题

消息队列核心概念：一条消息由谁保管，处理进度记在哪里 ​

订单创建之后，要发通知、更新搜索索引、生成统计报表。把这些工作放进消息系统，可以让下单请求先结束，后面的服务按自己的速度处理。但这也多出几个问题：通知服务停了，消息放在哪里？两个通知实例会不会各发一次？重启以后从哪里接着读？

本文围绕这条订单消息解释生产者、消费者、Broker、Topic、分区、消费组、偏移量和推拉模式。先分清各自的职责，后面才有办法讨论“不丢”“不重”和“有序”。产品依据采用 Kafka 4.2 与 RabbitMQ 4.2 的版本化文档；它们是本文的比较基线，不代表所有消息系统使用同一套机制。示例在 Go 1.27.2、Linux/amd64 运行，是没有网络、磁盘和故障恢复的确定性教学模型，不是 Broker 实测。

先看消息走过的四个地方 ​

角色或位置具体负责什么它不知道什么
生产者 Producer把业务事实或任务编码成消息，选择目的地，处理发送结果消费者是否已经完成业务，除非另有回执协议
Broker接收、保存或路由消息，按产品协议把消息交给消费者一笔数据库更新是否真正提交，除非应用向它报告
消费者 Consumer读取消息、校验内容、执行业务，然后确认处理进度生产者超时前是否已经成功发送过同一业务事件
业务存储或外部服务保存订单、账单、通知记录等最终效果Broker 的 ack 或 offset 是否提交成功

生产者和消费者是角色，不一定是两种独立进程。一个服务可以读入订单事件、处理后再发布账单事件；此时它同时扮演两种角色。Broker 也不等于一台机器，可能是一套有复制和协调机制的集群。多台机器能够容忍哪些故障，要继续看副本、确认和部署策略，单凭“有 Broker”不能得出可靠性结论。[1]

正在绘制图示…

可切换适合窗口或原始尺寸
查看图源代码
sequenceDiagram
    participant P as 订单服务
    participant B as Broker
    participant C as 通知服务
    participant D as 通知记录库
    P->>B: 发布 order.created，event_id=e1
    B-->>P: 生产端确认（按配置的接收条件）
    B->>C: 投递消息
    C->>D: 写入业务结果
    D-->>C: 提交成功
    C->>B: 消费确认或提交消费位置
放大图示

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码
sequenceDiagram
    participant P as 订单服务
    participant B as Broker
    participant C as 通知服务
    participant D as 通知记录库
    P->>B: 发布 order.created，event_id=e1
    B-->>P: 生产端确认（按配置的接收条件）
    B->>C: 投递消息
    C->>D: 写入业务结果
    D-->>C: 提交成功
    C->>B: 消费确认或提交消费位置

这张图里有两个方向不同的确认。生产端确认回答“Broker 对这次发布承担了什么责任”，消费端确认回答“这一消费者是否还需要这条消息”。RabbitMQ 的 publisher confirm 和 consumer acknowledgement 相互独立；两者都不直接证明另一端的业务结果。[2]

如果通知记录已经提交，最后一个确认却丢了，Broker 可能再次投递。反过来，消费者先确认、再写通知记录，进程可能在中间退出，留下“消息进度已前移、业务没有完成”的裂缝。理解这两个窗口，比先记一个“至少一次”的定义更有用。

Topic 是分类，分区才是日志的排队单位 ​

在 Kafka 的普通事件日志模型中，Topic 是一类记录的逻辑名字，例如 orders；一个 Topic 可以分成多个分区，每个分区各自追加记录、分配 offset。分区可以放在不同 Broker 上，于是读写能够并行。[1:1]

正在绘制图示…

可切换适合窗口或原始尺寸
查看图源代码
flowchart LR
    P["生产者：按 order_id 选分区"] --> P0["orders / P0<br/>offset 0: A-created<br/>offset 1: A-paid"]
    P --> P1["orders / P1<br/>offset 0: B-created<br/>offset 1: B-cancelled"]
    P0 --> C0["billing 组：实例 C0"]
    P1 --> C1["billing 组：实例 C1"]
    P0 --> A["audit 组：实例 A0"]
    P1 --> A
放大图示

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码
flowchart LR
    P["生产者：按 order_id 选分区"] --> P0["orders / P0<br/>offset 0: A-created<br/>offset 1: A-paid"]
    P --> P1["orders / P1<br/>offset 0: B-created<br/>offset 1: B-cancelled"]
    P0 --> C0["billing 组：实例 C0"]
    P1 --> C1["billing 组：实例 C1"]
    P0 --> A["audit 组：实例 A0"]
    P1 --> A

P0 的 offset 1 和 P1 的 offset 1 不是同一条消息，也不能据此判断谁先发生。记录的定位需要至少带上 Topic、分区和 offset。需要业务去重时，还要一个独立的 event_id：同一业务事件重发以后，可能获得新的 offset，物理位置并不能识别业务重复。

同一订单需要按创建、支付、退款的顺序处理时，常见做法是以订单 ID 作为分区键，让这些记录进入同一分区。这里有三个前提：生产者使用兼容的分区算法，分区数或路由规则没有悄悄改变，消费者没有把同一订单的处理随意并发化。Broker 给出分区内记录顺序，无法替应用保证多个 goroutine 的业务提交顺序。[1:2]

分区不是越多越好。它增加并行度，也增加日志、副本、调度和消费分配的管理成本。热点订单即使单独占一个分区，仍然需要串行处理它自己的状态变化。要扩容，先弄清是总吞吐不足，还是少数键造成热点。

RabbitMQ 的 queue 不能直接套进这张图 ​

RabbitMQ 常见的 AMQP 0-9-1 路径是“exchange 按 binding 路由到 queue，再由消费者取走”。Topic exchange 中的 topic 指路由匹配方式，不是 Kafka 的分区日志 Topic。普通 RabbitMQ queue 的消费者确认会让相应消息具备移除条件；Kafka 普通日志中的消费位置前移，不会要求删除那条日志。RabbitMQ 自己也有 Stream 产品形态，因此比较时必须说清队列类型。[3]

这会影响“通知和审计都要收到一份”的设计。Kafka 通常建立两个独立消费组；RabbitMQ 通常让交换机同时路由到两个队列，每个队列再放多个竞争消费者。给同一个 RabbitMQ 队列加第二个消费者，通常是在分担工作，不能据此实现每个服务各收一份。

消费组：分担一份工作，还是各自看一份日志 ​

普通 Kafka consumer group 把订阅分区分配给组内成员。在稳定分配下，一个分区由该组的一个成员负责；同一个分区可以同时由另一个组的成员读取。这样，通知服务的多个实例分担同一组工作，审计服务又有独立进度。[4]

想达到的效果Kafka 普通消费组怎么安排需要考虑的代价
增加通知处理能力同组增加成员，分担不同分区组内分区级并行度受可分配分区数限制
给审计服务一份完整事件流新建独立 group ID新组的起始位置和保留期必须匹配
重放历史修复索引使用隔离的新组或受控重置位置重放会再次触发处理，不能误发通知或重复扣款
某个成员退出后继续处理由组协调机制重新分配分区旧成员正在做的工作、未提交位置都要处理

成员加入、退出或分区变化时的重新分配叫 Rebalance。此时“我刚才负责 P0”不能当作永久所有权。旧任务如果继续写数据库，新成员又从已提交位置重读，就可能出现并行重复处理。正确性仍需业务幂等和适当的工作取消、排空或所有权检查。

这里特意写“普通消费组”：Kafka 4.2 还提供 Share Groups 这套队列式消费机制，按记录锁定与确认分担工作，不能拿普通组的分区独占规则概括它。选择 API 时先确认自己要日志消费还是任务分发。[5]

偏移量：读到了哪里，与完成到了哪里，是两回事 ​

Kafka 普通消费者至少有两种位置需要区分：当前读取位置，以及已经提交、可供恢复的位置。提交值通常表示下一条要读的 offset。例如完成 offset 10，应提交 11。实际客户端有的让你传“已处理记录”，由库内部加一；有的直接接收下一位置，必须查具体 API。[4:1]

下面把单分区日志压缩成一个内存模型。Fetch 故意只从已提交位置读取，不模拟真实客户端会自动推进的本地读取游标；它用来单独观察“恢复位置”。

go
l := NewLog("order-1", "order-2")
i, v, _ := l.Fetch("billing")
fmt.Println("billing:", i, v)
i, v, _ = l.Fetch("billing") // 尚未提交恢复位置
fmt.Println("without commit:", i, v)
if err := l.Commit("billing", i+1); err != nil { panic(err) }
i, v, _ = l.Fetch("billing")
fmt.Println("after commit:", i, v)
i, v, _ = l.Fetch("audit")
fmt.Println("audit:", i, v)

实际输出：

text
billing: 0 order-1
without commit: 0 order-1
after commit: 1 order-2
audit: 0 order-1

billing 前进，不会推动 audit；提交也没有删掉 order-1。但这个模型只保存内存，进程一退出所有位置都消失，不能用它论证真实产品的恢复能力。

更容易写错的是并行处理。假设已提交位置是 10，10、11、12 已经派给三个 worker。12 最先完成，如果直接提交 13，恢复时会跳过尚未完成的 10 和 11。应当推进的是“连续完成前缀”：

go
f := Frontier{Next: 10}
fmt.Println("finish 12 -> next:", f.Complete(12))
fmt.Println("finish 10 -> next:", f.Complete(10))
fmt.Println("finish 11 -> next:", f.Complete(11))
text
finish 12 -> next: 10
finish 10 -> next: 11
finish 11 -> next: 13

12 完成时只记住这个事实；10 完成后可以推进到 11；等 11 也完成，才越过早已完成的 12。文末模型使用单线程协调这个状态，真实多 worker 需要锁或一个专门的结果汇总 goroutine。真实 Kafka offset 可能因压缩、事务记录或过滤等原因不连续，不能把这里“每个整数都必须完成”的教学结构原样当作生产实现；生产跟踪应以实际派发的记录序列为准。[6]

推模式和拉模式：谁发起投递，谁限制在途数量 ​

方式谁发起拿消息常见用法主要边界
推送消费者订阅后，服务端持续发送RabbitMQ basic.consume要限制未确认数量，不能让慢消费者内存无限增长
拉取消费者请求一条或一批消息Kafka fetch；RabbitMQ basic.get请求频率、批量大小和等待时间会影响延迟与吞吐
长轮询拉取消费者发请求，服务端可等到有数据或超时Kafka fetch 等空闲时减少轮询；仍须处理超时和取消

Kafka 拉取模型适合让消费者自己掌握批量和速度。拉取不意味着每隔一毫秒空转一次；请求可以等待数据。推送也不意味着无法背压，RabbitMQ 手动确认结合 prefetch，可以限制尚未确认的投递数量。RabbitMQ 对 basic.get 不应用这个 prefetch 限制，所以两个 API 不能只换函数名就认为行为相同。[7][8]

还要分清协议和 Go API 的外观。SDK 后台可能主动拉取，前台却把结果交给一个回调或 channel。看到回调不能断言底层是推模式；看到 ReadMessage 也不能断言客户端没有提前缓存。

背压的目标是：处理不动时，明确停止接纳更多在途工作，或让上游按既定策略等待、限流、拒绝。下面用有界 channel 展示最小的接纳决策：

go
jobs := make(chan string, 2)
for _, job := range []string{"a", "b", "c"} {
    select {
    case jobs <- job:
        fmt.Println("accepted:", job)
    default:
        fmt.Println("backpressure:", job)
    }
}
text
accepted: a
accepted: b
backpressure: c

这里没有消费者运行，所以第三件工作明确未被接纳。真实系统不能只打印一行就丢掉它：可以暂停拉取、等待空位，或者在 API 边界返回可重试错误；如果消息已经从 Broker 取出，还要保留正确的未确认状态。把 channel 改得很大，只是把等待搬进内存，并没有提升处理能力。Go 有界 channel 的发送与阻塞规则见语言规范。[9]

最后，把排查问题的顺序记下来：先找消息保管在哪里，再找处理进度由谁保存，然后确认进度推进之前业务做完了什么。一个 offset、一个 ack、一个“发送成功”，只有放回这条链里才有完整含义。

依据与验证边界 ​

资料于 2026-10-13 核对。本文没有启动 Kafka 或 RabbitMQ，产品行为来自以下官方资料；本地实验只验证自建模型的状态变化和 Go channel 的接纳行为。

本地运行 go run .、go test -v ./...、go test -race ./...、go vet ./... 全部成功。TestCommitBounds 检查模型拒绝越界和后退提交，TestOutOfOrderCompletion 检查不能跨过未完成记录。这些测试不覆盖真实磁盘、网络、Rebalance，也不证明分布式系统“恰好一次”。

完整可复现代码 ​

把以下文件放在同一目录，使用 Go 1.27.2 执行：

sh
go run .
go test -v ./...
go test -race ./...
go vet ./...

go.mod ​

go.mod
module example.com/mq/core

go 1.27.2

main.go ​

代码块 · 79 行 · 点击收起 / 展开
go
package main

import "fmt"

// Log is a single-partition, append-only teaching model, not a broker.
type Log struct {
	records []string
	next    map[string]int
}

func NewLog(records ...string) *Log {
	return &Log{records: records, next: make(map[string]int)}
}

func (l *Log) Fetch(group string) (int, string, bool) {
	i := l.next[group]
	if i >= len(l.records) {
		return i, "", false
	}
	return i, l.records[i], true
}

func (l *Log) Commit(group string, next int) error {
	if next < l.next[group] || next > len(l.records) {
		return fmt.Errorf("invalid next offset: %d", next)
	}
	l.next[group] = next
	return nil
}

// Frontier tracks a contiguous completed prefix in an ordered worker batch.
type Frontier struct {
	Next int
	done map[int]bool
}

func (f *Frontier) Complete(offset int) int {
	if f.done == nil {
		f.done = make(map[int]bool)
	}
	if offset >= f.Next {
		f.done[offset] = true
	}
	for f.done[f.Next] {
		delete(f.done, f.Next)
		f.Next++
	}
	return f.Next
}

func demo() {
	l := NewLog("order-1", "order-2")
	i, v, _ := l.Fetch("billing")
	fmt.Println("billing:", i, v)
	i, v, _ = l.Fetch("billing")
	fmt.Println("without commit:", i, v)
	if err := l.Commit("billing", i+1); err != nil {
		panic(err)
	}
	i, v, _ = l.Fetch("billing")
	fmt.Println("after commit:", i, v)
	i, v, _ = l.Fetch("audit")
	fmt.Println("audit:", i, v)
	f := Frontier{Next: 10}
	fmt.Println("finish 12 -> next:", f.Complete(12))
	fmt.Println("finish 10 -> next:", f.Complete(10))
	fmt.Println("finish 11 -> next:", f.Complete(11))
	jobs := make(chan string, 2)
	for _, job := range []string{"a", "b", "c"} {
		select {
		case jobs <- job:
			fmt.Println("accepted:", job)
		default:
			fmt.Println("backpressure:", job)
		}
	}
}

func main() { demo() }

main_test.go ​

代码块 · 41 行 · 点击收起 / 展开
go
package main

import "testing"

func Example() {
	demo()
	// Output:
	// billing: 0 order-1
	// without commit: 0 order-1
	// after commit: 1 order-2
	// audit: 0 order-1
	// finish 12 -> next: 10
	// finish 10 -> next: 11
	// finish 11 -> next: 13
	// accepted: a
	// accepted: b
	// backpressure: c
}

func TestCommitBounds(t *testing.T) {
	l := NewLog("x")
	if err := l.Commit("g", 2); err == nil {
		t.Fatal("accepted an offset beyond the log")
	}
	if err := l.Commit("g", 1); err != nil {
		t.Fatal(err)
	}
	if err := l.Commit("g", 0); err == nil {
		t.Fatal("accepted a backwards commit in this model")
	}
}

func TestOutOfOrderCompletion(t *testing.T) {
	f := Frontier{Next: 10}
	for _, offset := range []int{12, 10, 12} {
		f.Complete(offset)
	}
	if f.Next != 11 {
		t.Fatalf("skipped incomplete 11: %d", f.Next)
	}
}

  1. Kafka 4.2 Introduction，Main Concepts and Terminology：生产者、消费者、Topic、分区与保留。 ↩︎ ↩︎ ↩︎

  2. RabbitMQ 4.2 Consumer Acknowledgements and Publisher Confirms：两个确认方向、连接丢失后的重投。 ↩︎

  3. RabbitMQ 4.2 Queues 与 Consumers：队列、竞争消费者、确认与顺序边界。 ↩︎

  4. Kafka 4.2 Consumer Configs 和 KafkaConsumer 4.2 API：group、position、committed position 与提交语义。 ↩︎ ↩︎

  5. Kafka 4.2 发布说明 与 Design，The Share Consumer：Share Groups 的不同消费模型。 ↩︎

  6. Kafka 4.2 Design，Log Compaction：压缩保留 offset，但日志可能有空洞。 ↩︎

  7. Kafka 4.2 Design，Push vs. pull：拉取、批量与等待。 ↩︎

  8. RabbitMQ 4.2 Consumer Prefetch 及 confirms 文档的 Prefetch and Polling Consumers：限制范围与 basic.get 例外。 ↩︎

  9. Go 语言规范，Send statements 与 Select statements。 ↩︎

保留各书来源与许可 · 示例运行条件以正文为准