第07章 · 点赞收藏与 Kafka 入门（正式版）

写在最前面：这一版和普通讲义有什么不同

这是《内容社区》课程第 07 章。前面六章我们一直在打地基：Go 语法、Gin + GORM、JWT 部署、接口抽象、SSO、发帖缓存。地基打完，这一章正式进入业务纵深。

但这一章有一个全文最大的难点：消息队列 Kafka。你前面没接触过它，所以我会花很大篇幅在「概念」上——先用"快递柜"这种生活类比让你"看见"它，再讲 Topic / Partition / 消费者组这些名词，最后才动手写代码。

这一版做了三件事：

1. 每一步都能跑。涉及外部依赖（Redis、Kafka）先给 Docker 启动命令，再写业务代码；每个示例都标文件路径、给预期输出。
2. 术语不跳过。凡是你没听过的词（消息队列、Topic、Partition、消费者组、Lua 脚本、软删除……）当场就解释，并配「补课胶囊」。
3. 结尾有面试问答。章末 16 道题，按"基础 → 进阶 → 追问"三档排列，每题告诉你一句话结论、展开怎么答、面试官想听什么词、新手最容易答错在哪。

如果你想要"高密度速查版"，可以看原版第 07 章 ；两版知识点完全一致，只是讲法不同。

一、本章导读

前面六章我们相继解决了"用户怎么登录"“文章怎么发"“怎么缓存提升性能"这些问题。但一个内容社区最热闹的地方，其实是互动——你刷到一篇好文章，会阅读、会点赞、会收藏。这些行为每秒可能发生几千次，如果直接往数据库里硬写，连接池会被打爆，页面会卡死。

上一章我们学会了用 Redis 给"发帖列表"做缓存，那是在读路径上做优化。这一章我们要在写路径上做优化：用 Redis + Lua 在内存里完成高频计数，把数据库的压力卸下来。同时，这一章会第一次正式引入消息中间件 Kafka——把"同步写库"改成"异步解耦”，发完请求立刻返回，落库和计数交给后台慢慢做。

学完本章你将拥有第一个"扛得住高并发互动"的服务，也会真正理解为什么后端工程里"消息队列"几乎是标配。下一章我们进入可观测性，把"出了问题怎么排查"补齐——而 Kafka 的 Rebalance、消费延迟等指标，正好是可观测性要监控的重点。

开读前先看一眼全局：点赞这个动作看着简单，但高并发下"每次点赞都直接写库"必挂。本章的解法是把点赞变成一条流水线——Web 层用 Lua 脚本快速判重后扔进 Kafka，消费者慢慢消化，进程内再用 BufferTrigger 攒批落库。下面这张图从左到右，就是一条点赞消息的完整生命周期。

flowchart LR
    A["用户点赞 / 收藏"] --> B["Web 层
Lua 脚本原子判重"]
    B -->|"发消息：削峰填谷"| C[("Kafka
like 主题")]
    C --> D["消费者组
Rebalance 分配分区"]
    D --> E["进程内聚合
BufferTrigger 攒批"]
    E --> F["批量写 Redis"]
    E --> G["批量写 MySQL"]
    D -.->|"同一 key 固定分区"| H["顺序消费
避免计数错乱"]
    D -.-> I["bitmap / set 判重
过滤重复消息"]

这张图怎么读：从左到右走一遍点赞消息的一生：请求先进 Web 层，由 Lua 脚本原子判重（防同一用户重复点赞刷数），通过后往 Kafka 的 like 主题发一条消息就立刻返回——这一步削峰填谷；消费侧由消费者组经 Rebalance 领走分区，消息先进进程内的 BufferTrigger 攒批，再分别批量写 Redis 与批量写 MySQL。两条虚线是保障支线：“同一 key 固定分区"保证顺序消费、计数不错乱，“bitmap / set 判重"再兜一道底过滤重复消息。

二、本章知识点大纲

先把这张地图看一遍，知道这一章要往哪走、走到哪算够。

必会（学完必须能独立写出来）

1. 用 biz + bizId 思路设计一个通用互动服务，让"文章 / 评论 / 动态"都能复用同一套点赞、收藏、计数逻辑，不重复造轮子。
2. 用 Redis Hash + Lua 脚本实现"检查 + 自增"的原子计数，说清为什么不能用两条命令分着写。
3. 写出 interactive / user_like_biz / collection / collection_item 四张表的建表语句，并解释联合唯一索引的作用。
4. 用 Docker Compose 把 Kafka 跑起来，用 Sarama 发一条消息、再起一个消费者组把它消费掉，并看到计数真的 +1。
5. 把"阅读计数"从同步写库改造成"领域事件 + 异步消费”，理解改造前后的链路差异。

理解（能讲清原理和取舍，不要求背代码）

1. 阅读 / 点赞 / 收藏三类互动的异同，以及为什么"合并成一个通用服务"是中小公司的高性价比方案。
2. Upsert 语义、软删除策略在互动业务里的应用，以及为什么"缓存一致性问题"在这里不需要解决。
3. Kafka 核心概念（Topic / Partition / Broker / Producer / Consumer / Consumer Group / Offset）各自的定位和关系。
4. 分区与消费者组的关系这条核心面试题：一个分区在同一时刻只能被组内一个消费者消费；消费者数 > 分区数会发生什么。
5. acks 参数（0 / 1 / -1）在"吞吐"和"不丢消息"之间的取舍，以及 ISR 在其中扮演的角色。

了解（知道有这回事，面试能说出名字和一句话定位）

1. Kafka 为什么快：顺序写磁盘 + 零拷贝（zero-copy）+ 批量压缩。
2. Rebalance（消费者组再均衡）的触发时机、代价，以及 Sarama 里 Setup / Cleanup 的调用时机。
3. Lua 脚本是什么、为什么 Redis 用它来保证原子性；软删除与硬删除的差异。
4. 消息有序性以"分区"为单位；全局有序与业务有序的取舍；Hash 分区器怎么用 key 保序。

前置依赖与补课清单

不需要你提前懂的：Kafka 用法、Docker Compose 写 Kafka、Sarama 客户端——本章从零讲。

预计时长与学习成果

• 预计时长：4 小时（精读 + 跟着敲所有代码 + 起 Kafka 跑通一条消息）；只求主干 2 小时可过一遍。
• 学完你能跑出来的东西（这是本章的验收标准，做到才算过关）：

  一个能对文章/评论做点赞、收藏、阅读计数的 InteractiveService；本地用 Docker 起好 Kafka，用 Sarama 发出一条 read_article 消息，消费者组消费它并把阅读数 +1，控制台能看到发送成功日志和计数变化。整套链路 go run 能跑通、能看到预期输出。

flowchart LR
    A["3.1 需求分析
三类互动异同"] --> B["3.2 阅读计数
Redis+Lua+Upsert"]
    B --> C["3.3 点赞
去重+软删除"]
    C --> D["3.4 收藏
1:N+事务"]
    D --> E["3.5 Kafka 入门
概念→Docker→实战"]
    E --> F["3.5.9 改造阅读计数
领域事件+异步"]
    F --> G["第四节 跟着写
分步实现"]
    G --> H["第七节 面试问答"]

这张图怎么读：这是本章的学习路线，箭头即先后顺序：A（3.1 需求分析，把三类互动统一成 biz+bizId 模型）→ B（3.2 阅读计数，Redis+Lua+Upsert 模板）→ C（3.3 点赞，去重+软删除）→ D（3.4 收藏，1:N+事务）——这四节都是同步写法；进入 E（3.5 Kafka 入门）后按"概念 → Docker → Sarama 实战"三步走，最后在 F（3.5.9）用领域事件把阅读计数改造成异步链路，收束到"第四节 跟着写"与"第七节 面试问答”。E→F 是本章难点所在，读到这里放慢速度。

三、正文

3.1 需求分析：阅读、点赞、收藏有什么不一样

先抛问题

内容社区几乎都有"阅读数 / 点赞 / 收藏"这三个功能。它们看起来像，但你在写代码前得先想清楚：它们真的是一回事吗？

• 阅读数：是不是只增不减？能不能"取消阅读”？
• 点赞：我点过赞之后，能不能再"取消"？系统怎么知道我点没点过？
• 收藏：我是不是可以有"技术"收藏夹、“美食"收藏夹，把同一篇文章放进不同夹子？

不把这些差异想清楚，你的数据表就会设计得一团糟：要么字段冗余，要么没法支持"取消点赞"“放进哪个收藏夹”。

给个直觉：书店的三种互动

把一家内容社区想象成一家书店，三种互动就像：

• 阅读 = “每有人进门就记一笔客流”，只增不减，不需要记"谁进来的”。
• 点赞 = “在书上贴便签，贴了能撕”，需要记录"谁贴过"，否则撕的时候不知道该撕哪张。
• 收藏 = “把书放进我的某个书架”，一个人可以有好几个书架（收藏夹），同一本书能放进不同书架。

三者形态不同，但本质都是**“某个用户对某个资源的某种计数动作”**——所以能抽象成同一个服务。

通用互动模型图

把三种互动抽象成一张图，核心是中间那个"统一服务"：

flowchart LR
    U[用户] -->|对某个资源做动作| A["资源 = biz + bizId
article/123, comment/456"]
    A --> I[InteractiveService
统一处理 阅读/点赞/收藏]
    I --> R[(Redis 计数缓存)]
    I --> DB[(MySQL 计数表)]

这张图怎么读：用户的动作（阅读 / 点赞 / 收藏）不直接对着某张表，而是先落在"资源 = biz + bizId"这个统一坐标上（如 article/123、comment/456），再交给中间的 InteractiveService 统一处理；服务内部向两个存储分发——Redis 计数缓存扛高频读，MySQL 计数表做最终落库。以后要支持评论点赞、动态点赞，只是换 biz 的取值，这条 U→A→I 的链路和 I→R / I→DB 的分发改都不用改。

展开讲：三类互动的差异

3.1.1 要不要做成「通用」功能

仔细想一下：评论可以被点赞，视频可以被点赞，动态也可以被点赞。如果每个业务各自实现一套点赞，代码重复严重，维护起来噩梦。

所以应该抽象出一个通用互动服务，用 biz + bizId 标识"哪个业务的哪条记录"：

• biz = "article"，bizId = 123 表示文章 id=123
• biz = "comment"，bizId = 456 表示评论 id=456

类似的设计在工业界很常见，也叫 resourceType + resourceId 或 requestType + requestId。只要换 biz，同一套代码就能服务评论、视频、动态，不用新写一套。

补课胶囊 · 什么是 biz 和 bizId 一句话：biz 是"业务类型"，bizId 是"这条记录在那个业务里的主键"，合起来唯一锁定一个资源。 类比：biz 像"小区名"，bizId 像"门牌号"，两者合起来才能找到具体一户。 最小示例：

// 文章 123 的阅读计数
IncrReadCnt(ctx, "article", 123)
// 评论 456 的点赞
Like(ctx, "comment", 456, uid)

回到主线：有了这对组合，后面所有的互动逻辑都围绕 (biz, bizId, uid) 转，接口签名再也不用为每种资源写一套。

3.1.2 拆分还是合并

阅读 / 点赞 / 收藏，是合成一个服务，还是拆成三个？一张表看清取舍：

本课程采用合并方案，这也是大多数中小公司的选择。

小收口

3.1 节解决的是"战略问题"：我们用 biz + bizId 把三类互动统一成一个 InteractiveService，用"合并方案"先活下来。接下来 3.2 ~ 3.4 逐个落地：阅读、点赞、收藏各自怎么存、怎么写。

3.2 阅读计数：高频计数业务的模板写法

先抛问题

阅读数是最高频的写操作——每打开一次文章 +1。如果每次都 SELECT 出当前值、内存 +1、UPDATE 写回，会发生什么？更关键的是：数据库扛不住这么高频的写，怎么办？

这一节给出"高频计数业务"的标准模板：MySQL 做最终落库（用 Upsert 原子自增），Redis 做高频读缓存，二者配合。

给个直觉：收银台与记账本

• MySQL 像"月底结账的总账本"，权威但翻起来慢，不适合每秒被写几千次。
• Redis 像"收银台旁随手记的小纸条"，写在内存里飞快，适合扛高频。
• 我们让"小纸条"先扛住高频读写，账本最后再慢慢对账——这就是"Redis 缓存 + MySQL 落库"的雏形。

3.2.1 设计思路

新建一个 InteractiveService，对外暴露 IncrReadCnt(biz, bizId) 方法。在文章详情接口的 Web 层聚合 ArticleService 和 InteractiveService。

为什么不在 GetPublishedById 内部直接 +1？因为"阅读"是一个独立的领域，未来还会有点赞、收藏等扩展，独立成服务更清晰，也方便后面用 Kafka 把它异步化（见 3.5.9 节）。

3.2.2 数据库设计要点

-- dao/interactive.sql
-- interactive 表：互动计数表，一张表同时存阅读/点赞/收藏三种计数
CREATE TABLE `interactive` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `biz` varchar(32) NOT NULL COMMENT '业务标识，如 article',
  `biz_id` bigint NOT NULL COMMENT '业务对象 ID',
  `read_cnt` bigint NOT NULL DEFAULT 0,
  `like_cnt` bigint NOT NULL DEFAULT 0,
  `collect_cnt` bigint NOT NULL DEFAULT 0,
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  -- 关键：联合唯一索引，保证同一 biz+bizId 只有一条记录
  UNIQUE KEY `biz_biz_id` (`biz`, `biz_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

一张图看清"阅读计数"的数据链路，以及为什么自增必须走数据库原子操作：

flowchart TD
    Req[文章详情请求] --> S[InteractiveService.IncrReadCnt]
    S --> C[Redis Hash 缓存 +1
扛高频读]
    S --> DB[MySQL Upsert
read_cnt = read_cnt + 1]
    DB -->|原子自增| Safe[行锁保证并发安全]

这张图怎么读：文章详情请求先打到 S（InteractiveService.IncrReadCnt），由它向两边分发：一边给 Redis Hash 缓存 +1（扛高频读），一边走 MySQL Upsert 落库；右侧"原子自增 → 行锁"这条边是重点——read_cnt = read_cnt + 1 由数据库行锁在一条语句内完成，并发安全；绝不能拆成"先 SELECT 再 UPDATE"两步，否则两个请求同时读到旧值会丢计数。

关键点 1：Upsert 语义。 你不知道是"首次插入"还是"记录已存在要 +1"，用 ON DUPLICATE KEY UPDATE（GORM 里是 clause.OnConflict）让数据库在一条语句里搞定：

// dao/interactive.go
// InsertOrUpdate 插入阅读计数；若 (biz, biz_id) 已存在则阅读数 +1
func (d *InteractiveDAO) InsertOrUpdate(ctx context.Context, biz string, bizId int64) error {
	// Upsert：利用联合唯一索引，冲突时执行 DoUpdates
	// 不能 "先 SELECT 再 +1 再 UPDATE"，会有并发问题
	// 直接让 MySQL 在一条语句里原子完成 read_cnt = read_cnt + 1
	return d.db.WithContext(ctx).Clauses(clause.OnConflict{
		Columns: []clause.Column{{Name: "biz"}, {Name: "biz_id"}},
		DoUpdates: clause.Assignments(map[string]interface{}{
			"read_cnt": gorm.Expr("read_cnt + ?", 1),
		}),
	}).Create(&Interactive{
		Biz: biz, BizId: bizId, ReadCnt: 1,
		Ctime: time.Now().UnixMilli(), Utime: time.Now().UnixMilli(),
	}).Error
}

Upsert 是什么：是 UPdate + inSERT 的合成词，意思是"存在就更新，不存在就插入"，一条语句搞定，避免你先查再决定是 insert 还是 update。GORM 用 clause.OnConflict 表达，底层就是 MySQL 的 ON DUPLICATE KEY UPDATE。

3.2.3 Redis 缓存设计

互动数据是高频访问数据，不做缓存数据库会被打爆。用 Redis 的 Hash（哈希）结构，一篇文章的三种计数放在同一个 key 下：

key:   interactive:article:123
field: read_cnt    -> 1024
field: like_cnt    -> 56
field: collect_cnt -> 12

为什么用 Hash 而不是 String？一篇文章的阅读 / 点赞 / 收藏在同一个 key 下，方便统一管理和批量读取（一次 HGETALL 全拿出来）。

3.2.4 补课胶囊：什么是 Lua 脚本，为什么 Redis 要用它保证原子性

一句话：Lua 是 Redis 内置的一门小脚本语言；把多条命令写进一个 Lua 脚本，Redis 会单线程、一次性、不被打断地执行完，从而保证"要么全做、要么不做"的原子性。

类比：你让前台"先看我有没快递、有就签收"，如果你分两句话对前台说（“看下有我快递没？"→ 前台说有 → “那帮我签收”），中间可能被别人插队；但如果你写张纸条把两件事一起交给前台，前台会一口气办完，别人插不进来。这张"纸条"就是 Lua 脚本。

最小示例（Redis 里直接跑）：

# 进入 redis-cli
redis-cli
# 定义一个 Lua 脚本：key 存在时才自增指定 field
EVAL "if redis.call('EXISTS', KEYS[1]) == 1 then return redis.call('HINCRBY', KEYS[1], ARGV[1], ARGV[2]) else return 0 end" 1 interactive:article:123 read_cnt 1

回到主线：阅读计数里"检查 key 是否存在 → 存在则自增"正是典型的 check-then-act，必须塞进一个 Lua 脚本，否则并发下会丢计数。

「检查 key 是否存在 → 存在则 HINCRBY」是典型的 check-then-act 场景，并发下会出问题。看代码：

// cache/interactive_redis.go
// luaIncrIfPresent：key 存在时把指定 field 自增 delta，不存在则返回 0
const luaIncrIfPresent = `
if redis.call("EXISTS", KEYS[1]) == 1 then
    return redis.call("HINCRBY", KEYS[1], ARGV[1], ARGV[2])
else
    return 0
end
`

// IncrReadCntIfPresent 缓存层的阅读计数自增（只在 key 已存在时）
func (r *InteractiveRedisCache) IncrReadCntIfPresent(ctx context.Context, biz string, bizId int64) error {
	key := r.key(biz, bizId)
	// ARGV[1] = field 名（read_cnt），ARGV[2] = 增量（1）
	_, err := r.client.Eval(ctx, luaIncrIfPresent, []string{key}, "read_cnt", 1).Result()
	return err
}

注意：HIncrBy 本身在 field 不存在时会自动置 0 再自增，所以脚本只判断 key 是否存在即可——只要 key 在，里面的 field 不存在也能安全自增。

3.2.5 缓存一致性问题：为什么不解决

时序问题：缓存没有数据时，读接口会回源数据库并回写缓存，同时写接口也在更新数据库。两者交错可能出现缓存和数据库不一致（比如缓存是旧值、数据库已是新值）。

结论：这里不需要解决。 理由有三：

1. 业务上阅读数少一点点不会有问题，用户根本感知不到。
2. 只有高并发文章才会出现一致性问题，而高并发文章本身阅读量就大，少几个无所谓。
3. 低并发文章几乎不会遇到这个并发场景。

小收口

3.2 节给出高频计数的标准模板：MySQL 用 Upsert 原子自增落库，Redis Hash 扛高频读，Lua 脚本保证"检查 + 自增"原子性，而"缓存一致性"在这里选择不解决（因为差几个无感知）。这套模板后面点赞、收藏的计数也照样用。

3.3 点赞：去重 + 计数

先抛问题

点赞比阅读多了一件事：要记录"某个用户是否点过赞”，否则无法判重（防止一个人狂点 1 万次把数刷爆），也无法"取消点赞"。而且"点赞 → 取消 → 再点赞"这种来回操作，数据库该怎么设计才高效？

3.3.1 数据库设计

-- dao/user_like_biz.sql
-- user_like_biz 表：记录哪个用户对哪个资源点过赞
CREATE TABLE `user_like_biz` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `biz` varchar(32) NOT NULL,
  `biz_id` bigint NOT NULL,
  `user_id` bigint NOT NULL,
  `status` tinyint NOT NULL DEFAULT 1 COMMENT '1 点赞 0 取消',
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  -- 联合唯一索引：同一个用户对同一个资源只有一条记录（点赞/取消都改这条）
  UNIQUE KEY `biz_biz_id_user_id` (`biz`, `biz_id`, `user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

这张表只存"谁对哪条资源点过赞"的关系，真正的总数在 interactive 表里（靠 like_cnt 字段）。两个表配合：关系表管"判重"，计数表管"总数"。

3.3.2 补课胶囊：软删除 vs 硬删除

一句话：硬删除是真的 DELETE 把行删掉；软删除是不删行，只把 status 字段改成"已删除/已取消"，数据仍在库里。

类比：硬删除像把纸条撕碎扔掉；软删除像在纸条上画个叉写着"作废"——纸条还在，只是标记为无效。

最小示例：

// 硬删除：取消点赞直接删行
db.Delete(&UserLikeBiz{}, "biz=? AND biz_id=? AND user_id=?", biz, bizId, uid)
// 软删除：只改状态，行还在
db.Model(&UserLikeBiz{}).Where(...).Update("status", 0)

回到主线：点赞/取消用软删除（改 status），性能好还能保留"谁点过赞"的历史数据。

考虑"点赞 → 取消 → 再点赞"的场景：

• 硬删除：取消时直接 DELETE，再点赞时 INSERT。简单，但删除 + 插入比 UPDATE 慢，且删掉后失去了"这个人历史点过赞"的数据。
• 软删除：用 status 字段标记 1/0，再点赞时只 UPDATE status=1。性能好，能保留数据。

实践建议：偏好软删除，因为性能好且保留数据，很多公司都这么干。

3.3.3 点赞 Service 实现

// biz/interactive.go
// Like 点赞：给资源总点赞数 +1，并标记该用户已点赞
func (s *interactiveService) Like(ctx context.Context, biz string, id int64, uid int64) error {
	// 步骤 1：给资源的总点赞数 +1
	err := s.repo.IncrLikeCnt(ctx, biz, id, 1)
	if err != nil {
		return err
	}
	// 步骤 2：标记该用户已点赞（status=true）
	return s.repo.MarkLiked(ctx, biz, id, uid, true)
}

// CancelLike 取消点赞：总点赞数 -1，并标记该用户未点赞
func (s *interactiveService) CancelLike(ctx context.Context, biz string, id int64, uid int64) error {
	err := s.repo.IncrLikeCnt(ctx, biz, id, -1)
	if err != nil {
		return err
	}
	return s.repo.MarkLiked(ctx, biz, id, uid, false)
}

DAO 层用 Upsert，因为"点赞 - 取消 - 再点赞"时记录已存在，需要更新 status 而不是再插一条：

// dao/interactive.go
// UpsertLike 插入或更新点赞关系（靠联合唯一索引去重）
func (d *InteractiveDAO) UpsertLike(ctx context.Context, biz string, bizId, uid int64, like bool) error {
	var st int8
	if like { st = 1 } else { st = 0 }
	return d.db.WithContext(ctx).Clauses(clause.OnConflict{
		Columns: []clause.Column{{Name: "biz"}, {Name: "biz_id"}, {Name: "user_id"}},
		DoUpdates: clause.Assignments(map[string]interface{}{
			"status": st,
			"utime":  time.Now().UnixMilli(),
		}),
	}).Create(&UserLikeBiz{
		Biz: biz, BizId: bizId, UserId: uid, Status: st,
		Ctime: time.Now().UnixMilli(), Utime: time.Now().UnixMilli(),
	}).Error
}

小收口

3.3 节解决"点赞怎么去重、怎么取消"。核心就两条：用关系表 user_like_biz 记录谁点过（联合唯一索引保证一人一条），用软删除的 status 字段表达点赞/取消（避免硬删除的慢与丢数据）。计数仍走 interactive 表的 Upsert 自增。

3.4 收藏：1:N 的收藏夹

先抛问题

收藏比点赞再复杂一点：一个用户有多个收藏夹（业务上叫"文件夹"），同一篇文章能放进不同夹子。这意味着收藏不是"一对一"的关系，而是"一个用户 ↔ 多个收藏夹 ↔ 里面有多篇文章"的结构。数据库该怎么建模？而且"插入收藏项"和"更新总收藏数"这两步必须同时成功或同时失败，怎么保证？

给个直觉：书架与书

• 用户 = 一个人
• 收藏夹（collection）= 他家的一个书架（“技术"架、“美食"架）
• 收藏项（collection_item）= 书架上的一本书

一个人可以有多个书架（1:N），一个书架上可以有多本书（1:N），所以整体是"用户 → 收藏夹 → 收藏项"的两层一对多。

3.4.1 表设计

-- dao/collection.sql
-- 收藏夹本体
CREATE TABLE `collection` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `name` varchar(128) NOT NULL,
  `user_id` bigint NOT NULL,
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  KEY `idx_user_id` (`user_id`)
);

-- 收藏夹与资源的关联（哪篇文章被收进了哪个夹）
CREATE TABLE `collection_item` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `collection_id` bigint NOT NULL,
  `biz` varchar(32) NOT NULL,
  `biz_id` bigint NOT NULL,
  `user_id` bigint NOT NULL,
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_collection_biz` (`collection_id`, `biz`, `biz_id`)
);

collection 和 collection_item 是 1:N 关系：一个收藏夹里有多个收藏项。

3.4.2 事务保证原子性

收藏时要同时做两件事：插入收藏项 + 更新总收藏数。任何一步失败，另一步也得回滚——否则会出现"计数加了但没记录"或"有记录但计数没加"的不一致。这就必须用数据库事务（transaction）把两步包成一个原子操作。

// dao/interactive.go
// AddCollectionItem 收藏：插入收藏项 + 更新总收藏数，两步必须原子
// 唯一键是 (collection_id, biz, biz_id)：同一篇文章可放进不同夹子，同一夹子内不重复收藏
func (d *InteractiveDAO) AddCollectionItem(ctx context.Context, biz string, bizId, uid, colId int64) error {
	// Transaction 开启一个事务，闭包返回 nil 才提交，返回 error 自动回滚
	return d.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		// 1. 插入收藏项（Upsert 防重：同一夹子内重复收藏则幂等跳过，DoNothing）
		itemRes := tx.Clauses(clause.OnConflict{
			Columns:   []clause.Column{{Name: "collection_id"}, {Name: "biz"}, {Name: "biz_id"}},
			DoNothing: true, // 冲突 = 同一夹子已收藏过，不覆盖、不报错
		}).Create(&CollectionItem{
			CollectionId: colId, Biz: biz, BizId: bizId, UserId: uid,
			Ctime: time.Now().UnixMilli(), Utime: time.Now().UnixMilli(),
		})
		if itemRes.Error != nil {
			return itemRes.Error // 返回 error → 整个事务回滚
		}
		// 重复收藏（冲突，RowsAffected=0）不增加总收藏数，幂等返回
		if itemRes.RowsAffected == 0 {
			return nil
		}
		// 2. 真正新增收藏项才自增 collect_cnt（Upsert 原子自增）
		return tx.Clauses(clause.OnConflict{
			Columns: []clause.Column{{Name: "biz"}, {Name: "biz_id"}},
			DoUpdates: clause.Assignments(map[string]interface{}{
				"collect_cnt": gorm.Expr("collect_cnt + ?", 1),
			}),
		}).Create(&Interactive{
			Biz: biz, BizId: bizId, CollectCnt: 1,
			Ctime: time.Now().UnixMilli(), Utime: time.Now().UnixMilli(),
		}).Error
	})
}

事务（Transaction）是什么：把若干条 SQL 打包成"一个要么全成、要么全败"的操作，靠数据库的 ACID 保证。在 GORM 里用 db.Transaction(func(tx *gorm.DB) error {...})，闭包返回 nil 就提交，返回 error 就回滚。收藏这种"两步必须同时成功"的场景，事务是标配。

小收口

3.4 节把"收藏"建模成两层一对多（收藏夹 → 收藏项），并用事务保证"插入收藏项"和"更新总收藏数"原子执行。到这里，阅读 / 点赞 / 收藏三大互动的存储与写法就齐了。在进 Kafka 之前，还有一块"读"的拼图要补上：文章详情页怎么把互动信息和文章内容一起高效返回。

3.4.3 查询接口：用 errgroup 并发聚合

先抛问题：文章详情页要同时展示"阅读/点赞/收藏总数"和"当前用户是否点赞/收藏”。这两块数据来自不同的 service（文章内容在 ArticleService，互动信息在 InteractiveService）。如果串行调用——先拉文章、再拉互动——接口延迟就是两者之和；能不能让它们同时去拉，谁慢等谁？

给个直觉：你去便利店买东西，店员一边帮你装袋、一边让同事去库房取缺的货，两件事并行做，你不用等其中一件做完才开始另一件。

Go 里用 errgroup 把"多个并发任务 + 等全部完成 + 任一出错就返回"打包好：

// handler/article_handler.go
// PubDetail 文章详情：并发拉文章内容 和 互动信息，再聚合返回
func (h *ArticleHandler) PubDetail(c *gin.Context) {
	// 从路径参数取文章 id
	id, err := strconv.ParseInt(c.Param("id"), 10, 64)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"msg": "参数错误"})
		return
	}
	// 从登录态取当前用户 uid（登录中间件已通过 c.Set("user_id", ...) 注入）
	uid := c.GetInt64("user_id")
	var (
		art  domain.Article
		intr domain.Interactive
	)
	// errgroup 把一组 goroutine 管起来：一个出错整体返回，并带 context 取消
	g, ctx := errgroup.WithContext(c.Request.Context())
	// 子任务 1：拉文章
	g.Go(func() error {
		var err error
		art, err = h.articleSvc.PublicById(ctx, id)
		return err
	})
	// 子任务 2：拉互动信息（含总数 + 当前用户是否点赞/收藏）
	g.Go(func() error {
		var err error
		intr, err = h.intrSvc.Get(ctx, "article", id, uid)
		return err
	})
	// Wait 等两个子任务都完成（或任一失败）
	if err := g.Wait(); err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"msg": "系统错误"})
		return
	}
	// 两个都成功，聚合返回
	c.JSON(http.StatusOK, gin.H{"article": art, "interactive": intr})
}

逐行讲解：

errgroup 是什么：是 golang.org/x/sync/errgroup 提供的工具，相当于"带错误传播和 context 取消的 WaitGroup”。前面章节用的 errgroup 就是它的升级版——比原生 sync.WaitGroup 多了"一个出错整体失败"的能力。

缓存取舍：只缓存"总数"（如 read_cnt），不缓存"个人是否点赞/收藏"。因为用户很少重复访问同一篇文章，缓存命中率极低，缓存它纯属浪费内存。是否缓存要基于业务理解判断，不是"能缓存就缓存"。

小收口

3.4 节补齐了"读"的拼图：文章详情用 errgroup 并发聚合文章内容与互动信息，延迟取最慢那路的 max 而非求和；缓存只放命中率高的"总数"，不放命中率低的"个人状态"。至此，互动服务的"读"与"写"两条线都齐了。下一节 3.5 是整个章节的"珠峰"——Kafka 消息队列入门。

3.5 Kafka 入门（本章最大难点）

前面的互动服务，所有写操作都是"同步"的——请求来了，你的代码当场连数据库、当场写、写完才返回响应。这在阅读量小的时候没问题；但阅读计数这种超高频写，如果每次请求都同步写库，数据库会被压垮，接口也会变慢。

解决思路是：把"记一笔阅读数"这件事，从"当场办"改成"丢给后台慢慢办"。中间那个"后台"就是消息队列。这一节我们从零讲 Kafka。

3.5.0 补课胶囊：消息队列是什么、为什么需要它

一句话：消息队列是一个"中转站"，生产者把消息丢进去就走，消费者按自己的节奏来取，双方不用同时在线。

类比（快递柜）：你（生产者）网购了一个包裹，不用等收件人（消费者）在家，直接把包裹放进小区快递柜（消息队列）的某个格口就走；收件人下班了自己去取。你俩从"必须当面交接"变成"通过柜子中转"，谁也不阻塞谁。

最小示例（Go 伪代码，先建立直觉）：

// 同步写法：调用方必须等数据库写完
IncrReadCnt(ctx, "article", id) // 可能要 50ms，接口被拖慢

// 异步写法：只把"要加阅读数"这件事丢进队列就返回
Produce("read_article", id) // 几毫秒搞定，落库交给消费者

回到主线：Kafka 就是这样一个"快递柜系统"。接下来我们讲它的三大价值和核心概念。

三大价值，每个配一个场景：

3.5.1 补课胶囊：Kafka 核心概念

一句话：Kafka 是一套"发布-订阅"式的分布式消息系统，下面这些名词是它的零件。

类比：把 Kafka 想象成一个物流中心——物流中心由若干栋仓库（Broker）组成，每栋仓库里分了若干个**分区（Partition）**当货架；商家（Producer）把货发到某个货架，快递员（Consumer）按组（Consumer Group）来取。

回到主线：先记熟这七个概念，后面所有图都围绕它们转。

一张架构图把上面七个概念串起来：

flowchart LR
    subgraph 生产者端
        P1[Producer A] --> T
        P2[Producer B] --> T
    end
    subgraph Kafka集群[Bootstrap Servers]
        T[Topic: read_article]
        T --> PA0[Partition 0]
        T --> PA1[Partition 1]
        T --> PA2[Partition 2]
        PA0 -. 落在不同 Broker .-> B1[(Broker 1)]
        PA1 -. 落在不同 Broker .-> B2[(Broker 2)]
        PA2 -. 落在不同 Broker .-> B3[(Broker 3)]
    end
    subgraph 消费者组
        PA0 --> C1[Consumer A]
        PA1 --> C2[Consumer B]
        PA2 --> C3[Consumer C]
    end
    C1 --> APP[业务处理
IncrReadCnt]
    C2 --> APP
    C3 --> APP

这张图怎么读：从左到右分三段看。生产者端：Producer A、Producer B 都把消息发往同一个 Topic（read_article），Topic 内部切成 Partition 0/1/2 三个分区，三条虚线表示分区尽量落在不同 Broker 上——这是副本与高可用的物理基础。消费端：注意箭头方向是分区指向消费者（PA0→Consumer A、PA1→Consumer B、PA2→Consumer C），体现"一个分区同一时刻只归组内一个消费者"；三个消费者各自处理完，最后汇到同一个业务处理（IncrReadCnt）。

Offset 关键点：消息在分区里是按顺序追加的，每条有一个递增的 offset。消费者处理完一批消息后，提交（commit）自己最新的 offset，Kafka 就知道"下次从这儿继续"。忘提交 offset 是新手最常犯的错（见 3.5.9 的坑）。

3.5.2 三大价值再落地：从图看异步 / 解耦 / 削峰

flowchart LR
    P[Producer 生产者] --> T[Topic 主题]
    T --> PA1[Partition 0]
    T --> PA2[Partition 1]
    T --> PA3[Partition 2]
    PA1 --> CG[Consumer Group 消费者组]
    PA2 --> CG
    PA3 --> CG
    CG --> APP[业务处理]

• 异步：Producer 把消息 Send 出去就返回，不用等 APP 处理完。
• 解耦：Producer 只管发到 read_article 这个 topic，它根本不知道后面有几个消费者、干什么用。
• 削峰：Producer 瞬时发 10 万条，topic 先把它们攒在分区里，消费者按自己的速度（比如每秒 1 万）慢慢消费，不会把下游打挂。

3.5.3 核心面试题：分区与消费者组的关系

一张图说明"一分区一消费者"和消息积压的关系：

flowchart TD
    T[Topic 有 3 个分区] --> P0[P0]
    T --> P1[P1]
    T --> P2[P2]
    P0 --> C1[消费者 A]
    P1 --> C2[消费者 B]
    P2 --> C3[消费者 C]
    D[消费者 D 分到 0 个分区] -. 多余消费者闲置 .-> P0

关键规则：一个分区在同一时刻只能被同一个消费者组内的一个消费者消费。

由此衍生两条铁律：

• topic 有 N 个分区 → 消费者组最多 N 个消费者有效，多出来的会闲置（分不到分区，干瞪眼）。
• 想加快消费速率不能无限加消费者，要先加分区——否则新增的消费者拿不到分区，纯浪费。

这就是「消息积压」问题的根因：消费者不够快，分区又不够多，消息在队列里越堆越多。

再补一个理解消费者组的关键视角：一个消费者组 = 一个关心这个 topic 的业务方。比如"用户注册"这件事，搜索团队关心新用户（要建索引），推荐团队也关心新用户（要做冷启动推荐）——那么搜索自己是一个消费者组，推荐是另一个消费者组。组与组之间互不影响，各自都能拿到全量消息；只有组内才需要分摊分区、避免重复消费。所以"不同消费者组可以同时消费同一个分区"和"同一消费者组内一个分区只给一个消费者"并不矛盾：前者是业务方之间各拿各的，后者是同一个业务方内部别重复干活。

延伸：消息积压怎么办？ 既然根因是"分区数 < 需要的消费能力"，工业界三招：

注意：如果是"有序消息积压"（如同一用户的行为必须按序处理），异步消费时也要保证哈希路由，让同一 key 的消息进同一个 goroutine，否则并发会打乱顺序。

3.5.4 消息有序性：以分区为单位

Kafka 的有序性以分区为单位保证：一个分区内，消息按写入顺序被消费。

• 全局有序：只用一个分区（牺牲并发，吞吐上不去）。
• 业务有序：用 Hash 分区器，把同一业务 key 的消息路由到同一分区（既保住需要的顺序，又保留并发）。

// producer/config.go
// 用 Hash 分区器保证"业务有序"：同 key 的消息一定进同一个分区
config := sarama.NewConfig()
config.Producer.Partitioner = sarama.NewHashPartitioner // 按 key 哈希
// 发送时设置 key，比如 biz_id，这样同一篇文章的阅读事件都在同一分区
msg := &sarama.ProducerMessage{
	Topic: "read_article",
	Key:   sarama.StringEncoder(strconv.FormatInt(bizId, 10)), // key 决定分区
	Value: sarama.ByteEncoder(payload),
}

3.5.5 acks 参数：可靠性核心

生产者发消息时 acks 有三个取值，本质是"到底要等谁确认，我才算发送成功"：

核心问"谁 ack 才算数"：

• 0：TCP 层发出去就算（网络层 ack）
• 1：主分区 ack
• -1：所有 ISR 都 ack

3.5.6 Docker 启动 Kafka（新版无需 ZooKeeper）

新版 Kafka 已把元数据管理内置（KRaft 模式），不需要再起一个 ZooKeeper。下面这份 docker-compose.yaml 直接能跑：

# docker-compose.yaml
# 用 bitnami 的 Kafka 镜像，KRaft 模式，不需要 ZooKeeper
services:
  kafka:
    image: bitnami/kafka:latest
    ports:
      - "9092:9092"
    environment:
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER

启动命令：

# 在当前目录（有 docker-compose.yaml 的地方）
docker compose up -d

# 看一眼是否起来
docker compose ps

预期输出（关键行）：

NAME                IMAGE                 STATUS
xxx_kafka_1         bitnami/kafka:latest  Up (healthy)

如果你的机器没有 docker compose（老版本是 docker-compose），用对应命令即可。起好之后 Kafka 在 localhost:9092 监听，后面 Go 代码连这个地址。

3.5.7 Go 客户端选型

课程选 Sarama，因为出去工作遇到它的概率最大。

3.5.8 补课胶囊：为什么 Kafka 能这么快

一句话：Kafka 快，靠的是"顺序写磁盘"“零拷贝"“批量 + 压缩"三板斧，而不是"因为用了内存”。

类比：机械硬盘像磁带，随机读写要反复移动磁头（慢）；顺序写像流水线上一直往前推（快）。Kafka 把消息顺序追加到磁盘日志文件，写盘速度反而比随机写内存数据库还快。

回到主线：知道这三点是"了解"级别，面试能说出名字即可。

• 顺序写磁盘：消息只往日志末尾追加（append-only），避免磁盘随机寻道。
• 零拷贝（zero-copy）：消费者读消息时，数据从磁盘文件经 sendfile 直接拷到网卡，跳过内核态 ↔ 用户态的多次拷贝。
• 批量压缩：生产者把多条消息攒成一批再压缩发送，网络传输量大幅减少。

3.5.9 用 Kafka 改造阅读计数：领域事件 + 异步消费

原方案的痛点：原来的 GetPublishedById 接口里同步调用 IncrReadCnt 写库。读文章是高频操作，原本接口只有数据库读，现在多了一次写，DB 压力陡增，接口也变慢。

改造后的链路：

flowchart LR
    Req[用户读文章] --> H[ArticleService]
    H -->|当场返回文章| U[用户]
    H -->|异步投纸条| K[(Kafka
read_article)]
    K --> C[消费者组]
    C --> S[InteractiveService.IncrReadCnt]
    S --> DB[(MySQL)]

引入 DDD 里的「领域事件」概念：用户读了文章，发出一个 ReadArticleEvent，对原接口透明。

生产者侧（文章服务发消息）：

// domain/event.go
// ReadArticleEvent 阅读领域事件：用户读了哪篇文章
type ReadArticleEvent struct {
	Uid int64 // 谁读的
	Aid int64 // 读了哪篇
}

// biz/article.go
// PublicById 读文章接口：返回文章，并异步发阅读事件
func (s *articleService) PublicById(ctx context.Context, id, uid int64) (domain.Article, error) {
	art, err := s.repo.PublicById(ctx, id)
	if err != nil {
		return art, err
	}
	// 异步发送领域事件，不阻塞主流程
	evt := ReadArticleEvent{Uid: uid, Aid: id}
	payload, _ := json.Marshal(evt)
	// 用文章 id 当 key → 同一篇文章的阅读事件进同一分区，保业务有序
	_ = s.producer.Produce(ctx, "read_article", strconv.FormatInt(id, 10), payload)
	return art, nil // 用户立刻拿到文章，记阅读数这事甩给后台
}

消费者侧（订阅 topic，调用计数）：

// mq/read_article_handler.go
// ConsumeClaim 核心消费逻辑：从分区里一条条取消息
func (h *readArticleHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		var evt ReadArticleEvent
		_ = json.Unmarshal(msg.Value, &evt)
		// 步骤 1：业务处理（落库 +1）
		_ = h.intrSvc.IncrReadCnt(context.Background(), "article", evt.Aid)
		// 步骤 2：业务成功后提交偏移量，否则重启会重复消费
		sess.MarkMessage(msg, "")
	}
	return nil
}

⚠️ 新手必踩的坑：忘记 MarkMessage 会重复消费。只有调用 MarkMessage 提交 offset，Kafka 才认为这条消息"消费过了”。如果处理成功却没提交，消费者重启后会从旧 offset 重新拉，这条消息被再处理一次。所以"先确保业务成功，再 MarkMessage"是铁律。

后续要加新功能（比如"阅读记录"给用户看历史），只要再起一个消费者组订阅同一个 topic 即可，对原接口完全无侵入。

3.5.10 副本机制：主分区、从分区与高可用

前面说"topic 有 3 个分区"指的是 3 个主分区（Leader）；每个主分区还会有若干从分区（Follower）做副本（常见配置一主两从）。

• 写入：生产者的消息只写主分区，从分区不断从主分区同步。
• 高可用：某 broker 宕机时，ISR 里数据最全的从分区会被提升为新主分区，服务不中断、数据不丢。

flowchart TD
    P[Producer] -->|只写主分区| L0[Partition0 主]
    L0 -->|同步| F0A[Partition0 从 A]
    L0 -->|同步| F0B[Partition0 从 B]
    L0 --> C[Consumer 只读主分区]

分区分布两条规则：

• 同一 topic 的不同主分区，尽量不落在同一个 broker 上。
• 同一个分区的主、从副本，必须落在不同的 broker 上。

目标：任何一个 broker 崩溃，最多导致 topic 某一个主分区需要换主，不至于整个 topic 不可用。

ISR（In Sync Replicas） = “跟得上主分区节奏的从分区集合”。从分区如果因网络/磁盘慢长期落后，会被踢出 ISR。acks=-1 时要等所有 ISR 成员都确认写入才返回成功——这就是"不丢消息"的底气：即使主分区所在机器立刻宕机，数据在 ISR 从分区上还有一份。

3.5.11 Rebalance：消费者组的再均衡

**Rebalance（再均衡）**是消费者组内"分区重新分配"的过程。白话：组里的成员或分区数变了，Kafka 重新商量"谁消费哪几个分区"。

触发时机：

• 消费者加入或退出（扩容、发版重启、消费者崩溃）
• 消费者被判定死亡（心跳超时）
• topic 分区数变化
• 订阅的 topic 正则匹配到了新 topic

Rebalance 的代价：

1. 期间整组停止消费：分区所有权要重新分配，正在消费的分区会被暂停。
2. 可能重复消费：分区被移交时，上一个消费者处理了但还没提交 offset 的消息，会被新消费者再消费一次。

对应到 Sarama：Setup 在 rebalance 完成后、开始消费前调用，Cleanup 在消费结束、下一次 rebalance 前调用——所以这两个方法适合做"初始化 / 清理"，不适合放业务逻辑。

减少 rebalance 的实践：

• 心跳与会话超时留够余量，别让一次 Full GC 就被 Kafka 判定死亡。
• 消费逻辑里不要做超长阻塞的活，凑批一定要设超时。
• 发版时主动 cancel context 优雅退出，别让 Kafka 靠心跳超时去猜你死了。

3.5.12 分区器全家福 & Kafka Shell 工具

Sarama 支持多种分区器，一次看全：

Docker 起来后，进容器用命令行工具调试（验证 topic 建好没、消息发没发出去）：

# 进容器（容器名可能不同，用 docker ps 看）
docker exec -it $(docker ps -qf "name=kafka") /bin/sh

# 在容器内切到 Kafka 的 bin 目录（bitnami 镜像路径）
cd /opt/bitnami/kafka/bin

# 创建 topic：3 个分区，副本因子 1（本地练习够用）
kafka-topics.sh --bootstrap-server localhost:9092 --topic first_topic \
  --create --partitions 3 --replication-factor 1

# 查看 topic 详情：分区数、副本分布、ISR
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic first_topic

# 起一个命令行消费者，实时盯着这个 topic 的消息
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first_topic

# 另开终端起一个命令行生产者，手动发一条消息
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic first_topic

记不住命令非常正常——这些命令老师自己也记不住，要用的时候看文档抄即可。Sarama 也自带 Go 命令行工具：go install github.com/IBM/sarama/tools/kafka-console-consumer@latest 装一个即可。

3.5.13 Sarama 实战：生产者与消费者组

同步生产者：

// mq/producer.go
// NewSyncProducer 创建同步生产者（发完等 Kafka 确认才返回）
func NewSyncProducer(addrs []string) (sarama.SyncProducer, error) {
	config := sarama.NewConfig()
	// acks 选择：-1 最安全，1 平衡，0 最快
	config.Producer.RequiredAcks = sarama.WaitForAll // 等价 acks=-1
	config.Producer.Return.Successes = true          // 同步发送必须开启
	return sarama.NewSyncProducer(addrs, config)
}

// Produce 发送一条消息，用 key 保证业务有序
func Produce(p sarama.SyncProducer, topic, key, value string) error {
	msg := &sarama.ProducerMessage{
		Topic: topic,
		Key:   sarama.StringEncoder(key),  // 同 key 进同分区
		Value: sarama.ByteEncoder(value),
	}
	_, _, err := p.SendMessage(msg)
	return err
}

异步生产者（适合高吞吐，往 channel 丢就走）：

// mq/async_producer.go
// NewAsyncProducer 创建异步生产者
func NewAsyncProducer(addrs []string) (sarama.AsyncProducer, error) {
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true // 想拿成功通知就开
	config.Producer.Return.Errors = true    // 想拿错误通知就开
	return sarama.NewAsyncProducer(addrs, config)
}

// runAsyncProducer 监听成功/失败两个 channel
func runAsyncProducer(p sarama.AsyncProducer) {
	p.Input() <- &sarama.ProducerMessage{Topic: "test", Value: sarama.StringEncoder("hi")}
	for {
		select {
		case msg := <-p.Successes():
			log.Printf("发送成功 partition=%d offset=%d", msg.Partition, msg.Offset)
		case err := <-p.Errors():
			log.Printf("发送失败 %v", err)
		}
	}
}

消费者组（实现 ConsumerGroupHandler 接口）：

// mq/read_article_handler.go
// readArticleHandler 实现 sarama.ConsumerGroupHandler
type readArticleHandler struct {
	svc InteractiveService // 注入业务 service
}

// Setup 在 rebalance 之后、消费之前调用
func (h *readArticleHandler) Setup(sarama.ConsumerGroupSession) error { return nil }

// Cleanup 在消费结束、rebalance 之前调用
func (h *readArticleHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }

// ConsumeClaim 核心消费逻辑
func (h *readArticleHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		err := h.svc.OnReadEvent(context.Background(), msg.Value)
		if err != nil {
			// 失败处理：重试 / 死信队列 / 记录日志后跳过
			log.Printf("消费失败 topic=%s partition=%d offset=%d err=%v",
				msg.Topic, msg.Partition, msg.Offset, err)
		}
		// 阅读计数属"可丢"场景：无论成功失败都标记消费完成并提交偏移量（失败仅记日志跳过）
		sess.MarkMessage(msg, "")
	}
	return nil
}

启动消费者，用 context 控制退出：

// mq/consumer.go
// 用 context 控制消费者组的生命周期，避免 goroutine 泄漏。
// 注意：ctx 必须由调用方（如 main）持有并传入，在真正停机时才 cancel。
// 不能在这个"启动后立即返回"的函数里 defer cancel()——否则 goroutine 一启动
// ctx 就被取消，Consume 立刻返回 context.Canceled，消费者一条消息都消费不到。
func RunConsumer(ctx context.Context, cg sarama.ConsumerGroup) {
	go func() {
		for {
			// Consume 阻塞消费，出错或 ctx 取消会返回
			err := cg.Consume(ctx, []string{"read_article"}, &readArticleHandler{})
			if err != nil {
				log.Println(err)
			}
			if ctx.Err() != nil {
				return // 主动退出
			}
		}
	}()
}

从指定偏移量开始消费（重放历史消息）：可在 Setup 里重置起始 offset：

func (h *readArticleHandler) Setup(sess sarama.ConsumerGroupSession) error {
	// sarama.OffsetOldest = 从该分区最早的消息开始；也可填具体 offset 数字
	sess.ResetOffset("read_article", 0, sarama.OffsetOldest, "")
	return nil
}

3.5.14 批量消费提升吞吐

读文章特别适合批量处理：生产者一次发一条，消费者攒一批一起写库，把"100 次事务提交"变成"1 次"，性能差一个数量级。一张图看清差别：

flowchart LR
    subgraph 单条["单条消费"]
        M1[每条消息] --> W1[一次事务写库]
    end
    subgraph 批量["批量消费"]
        B1[攒 100 条] --> W2[一次事务批量写库]
    end

// mq/batch_handler.go
// batchReadHandler 攒批消费：攒够 batch 条或等 wait 超时，一次性刷库
type batchReadHandler struct {
	svc   InteractiveService
	batch int           // 一批大小，比如 100
	wait  time.Duration // 凑批超时，比如 1s
}

func (h *batchReadHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	batch := make([]int64, 0, h.batch)
	timer := time.NewTimer(h.wait)
	defer timer.Stop()
	for {
		select {
		case msg, ok := <-claim.Messages():
			if !ok {
				return nil
			}
			var evt ReadArticleEvent
			_ = json.Unmarshal(msg.Value, &evt)
			batch = append(batch, evt.Aid)
			// 先提交位点再攒批：仅适用于"可丢"的计数场景（宁可少记几条也不阻塞队列）。
			// 代价：进程在 flush 前崩溃/失败时，这一批消息不会被重放、计数会少。
			// "不能丢"的业务必须反过来——flush 成功后再 MarkMessage（见 3.5.9 节的铁律）
			sess.MarkMessage(msg, "")
			if len(batch) >= h.batch {
				h.flush(batch)
				batch = batch[:0]
				timer.Reset(h.wait)
			}
		case <-timer.C:
			if len(batch) > 0 {
				h.flush(batch)
				batch = batch[:0]
			}
			timer.Reset(h.wait)
		}
	}
}

// flush 一次事务批量更新多条阅读计数
func (h *batchReadHandler) flush(aids []int64) {
	_ = h.svc.BatchIncrReadCnt(context.Background(), "article", aids)
}

DAO 层用一条事务批量更新：

// dao/interactive.go
// BatchIncrReadCnt 一次事务批量 Upsert 多条阅读计数
func (d *InteractiveDAO) BatchIncrReadCnt(ctx context.Context, biz string, ids []int64) error {
	return d.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		for _, id := range ids {
			err := tx.Clauses(clause.OnConflict{
				Columns: []clause.Column{{Name: "biz"}, {Name: "biz_id"}},
				DoUpdates: clause.Assignments(map[string]interface{}{
					"read_cnt": gorm.Expr("read_cnt + ?", 1),
				}),
			}).Create(&Interactive{Biz: biz, BizId: id, ReadCnt: 1,
				Ctime: time.Now().UnixMilli(), Utime: time.Now().UnixMilli()}).Error
			if err != nil {
				return err
			}
		}
		return nil
	})
}

小收口

3.5 节是整章的珠峰：我们先用"快递柜"理解消息队列的三大价值，再认识 Kafka 的七个核心概念（Topic/Partition/Broker/Producer/Consumer/Consumer Group/Offset），然后落地了分区与消费者组的关系（核心面试题）、acks 可靠性、副本与 ISR、Rebalance、分区器、Docker 启动、Sarama 生产者/消费者组实战，最后用"领域事件 + 异步消费 + 批量消费"把阅读计数改造成高吞吐链路。

3.6 进阶：进程内流量聚合器——自研 BufferTrigger

先抛问题。3.5.14 的攒批解决了"消费者侧"的批量，但还有一段路没优化：Web 进程里，每个"阅读 +1"请求都要独立发一次 Kafka 消息（或一次 Redis 写）。瞬时 10 万 QPS 就是 10 万次网络往返——写放大发生在数据进缓冲之前。能不能在进程内先把零散请求"攒一攒"再统一输出？

给直觉。这就是流量聚合器 BufferTrigger 的活儿（Java 课程里是现成的开源工具，Go 生态没有同等成熟的中间件，正好自己写一个）。它只有三个要素：

// pkg/buffer/buffer_trigger.go —— 极简版 BufferTrigger（Go 自研思路）
type BufferTrigger[T any] struct {
    mu      sync.Mutex
    buf     []T
    maxCnt  int           // 数量阈值：攒够就刷
    onFlush func([]T)     // 输出函数：批量写 Redis / 发 Kafka
}

func (b *BufferTrigger[T]) Add(v T) {
    b.mu.Lock()
    b.buf = append(b.buf, v)
    full := len(b.buf) >= b.maxCnt
    batch := make([]T, 0)
    if full {
        batch, b.buf = b.buf, make([]T, 0, b.maxCnt)
    }
    b.mu.Unlock()
    if full { // 在锁外执行 flush，别让批量输出拖住 Add
        b.onFlush(batch)
    }
}

// time.Ticker 补上时间阈值：每 200ms 把攒不满的一批也刷出去
func (b *BufferTrigger[T]) Start(ctx context.Context) {
    t := time.NewTicker(200 * time.Millisecond)
    go func() {
        for {
            select {
            case <-t.C:
                b.mu.Lock()
                batch := b.buf
                b.buf = make([]T, 0, b.maxCnt)
                b.mu.Unlock()
                if len(batch) > 0 {
                    b.onFlush(batch)
                }
            case <-ctx.Done():
                b.flushRemaining() // 优雅停机：把剩余数据刷完再退出
                return
            }
        }
    }()
}

它顺带解决了两个问题：

1. 并发写放大：100 个 goroutine 同时 +1，聚合后只有 1 次批量操作，Redis / DB 的 QPS 直接降一到两个数量级；
2. 划清"模糊计数 vs 精准计数"的分界：阅读量、浏览量这类能容忍短暂不一致的指标走缓冲区批量刷（模糊计数），性能拉满；资金、库存这类必须精准的场景不能走缓冲，要么同步写、要么用 MQ 事务消息 / 本地消息表保证必达（第 15 章支付的本地消息表就是精准路线）。

常见坑：进程重启或发布时，缓冲区里没刷出去的数据会丢——所以必须有 flushRemaining 的优雅停机钩子；另外"模糊"的底气来自"差几个无所谓"，千万别把精准场景混进缓冲区。

小收口：还有一个杀手锏——爆火内容的延迟消息削峰。某明星官宣的瞬间计数洪峰灌进来，可以用 MQ 的延迟消息把"非紧急计数"推迟几秒错峰消费（Kafka 原生不支持延迟，RocketMQ 有；Kafka 侧可以在消息里带时间戳、消费者判断"不到点先攒着"模拟），极端场景甚至直接丢弃部分请求——大多数计数丢了也没人发现。削峰 + 聚合 + 可降级，是计数写入的完整三板斧。顺带说一道计数表的设计选择题：给 interactive 表"加列"（read_cnt / like_cnt / collect_cnt，本课程的方案——类型固定、查询快，但新增计数类型要改表）还是"行式"存 (biz, biz_id, count_key, count_value)（不用改表、Redis 天然配 hash，但一行变多行）——计数类型稳定选前者，扩展频繁选后者。而数据量大到要分库分表时注意：别用 biz_id（内容 id）做分片键——爆火内容会让某个分片成为单点热点，按用户 id 分更均匀。

3.7 顺序消费：全局顺序 vs 局部顺序

先抛问题。Kafka 一个 topic 有多个分区，消费者组又可能并发处理——那"先关注、后取关"两条消息，会不会被消费成"先取关、后关注"？计数 +1 和 -1 会不会错序？

给直觉。先把概念掰开：

• 全局顺序：整个 topic 只用 1 个分区、1 个消费者，所有消息绝对有序——代价是吞吐只剩单分区水平，生产上基本不用；
• 局部顺序：按业务键（如 userId）hash 到固定分区，同一用户的消息永远落在同一分区，分区内 FIFO、分区间并行。吞吐和有序兼得，这是工业界标准答案。

// 生产者侧：按 userId 选分区（sarama 的 hash 分区器默认行为）
// key 相同 → 分区相同 → 分区内先进先出
msg := &sarama.ProducerMessage{
    Topic: "follow-events",
    Key:   sarama.StringEncoder(strconv.FormatInt(uid, 10)), // 关键：业务键做 key
    Value: sarama.ByteEncoder(data),
}

关键判断：关注/取关需要顺序吗？ 答案是不需要。用户 A"关注 → 取关"和"取关 → 关注"的最终状态都是"未关注"，计数一个 +1-1、一个 -1+1，最终都是 0——这种场景只要最终一致，错序无所谓。真正需要局部顺序的是"同一对象的多次变更必须按先后生效"：比如同一篇内容的计数变更按 biz_id 分区（第 18 章 Canal 按主键 hash 分区保序是同一个思想），再比如同一用户的状态机流转。

常见坑：

1. 分区数变更破坏顺序：从 3 分区扩到 6 分区，hash 路由会变，扩容瞬间新旧分区里都可能有"同一用户"的消息——分区规划要提前留量（呼应 3.5.3 积压三招里"分区定了再改有成本"）；
2. 重试破坏顺序：消费者失败重试如果丢进重试 topic，后面的消息先处理了，顺序就乱——局部顺序场景要么同步阻塞重试，要么接受"重试批次最终一致"。

小收口：顺序消费口诀——全局顺序牺牲吞吐，局部顺序按业务键 hash 分区；先问业务要不要顺序，大多数"计数 / 关注"场景最终一致就够。

3.8 进阶：用 bitmap 判断"是否点赞"——内存从 GB 到 MB

先抛问题。3.3 节的判重靠 user_like_biz 表的唯一索引，“这个用户赞过没有"要查一次库（或缓存一份）。但如果产品要求"文章详情页每次加载都展示是否点赞”，亿级用户 × 百万内容的高频查询，关系表 + 缓存的方案内存扛得住吗？

给直觉。换一把更省的刀：Redis bitmap（位图）。原理朴素到极致——一个字节有 8 个 bit，每个 bit 是 0 或 1，正好表示"没点赞 / 点赞了"：

• key = liked:{biz}:{bizId}（某条内容的位图）
• offset = userId（该用户在位图里的位置）

// 点赞：把 offset=uid 的位设为 1（取消点赞设为 0）
rdb.SetBit(ctx, "liked:article:101", int(uid), 1)
// 判断是否点赞：GETBIT 一次内存操作，O(1)
v, _ := rdb.GetBit(ctx, "liked:article:101", int(uid))
liked := v == 1
// 位为 1 的总数：BITCOUNT（注意它是"当前为 1 的个数"，展示用的点赞数仍以计数表 like_cnt 为准）

内存账要会算（面试必考）：1 亿用户需要一个 1 亿 bit 的位图 = 1 亿 / 8 / 1024 / 1024 ≈ 11.9 MB。对比用 Set 存 1 亿个 int64 userId（每个 8 字节再加指针开销，好几个 GB），压缩了一千倍。但注意位图按 offset 上限分配内存：如果 userId 是雪花 ID 那种天文数字，直接当 offset 会把位图撑爆——要么用自增 uid（位图友好），要么加一层"uid → 序号"的映射。

升级与局限。如果业务只需要"见过没"的存在性判断、能容忍极小误判，可以进一步换布隆过滤器（多个 hash 函数映射多个 bit，结论是"一定不存在 / 可能存在"）——布隆的原理和取舍 06 章 3.22 节讲过，这里是它第二次出场：判断"是否点赞"它更省内存，代价是无法删除单个元素，所以"点赞 / 取消"来回切的场景只能用 bitmap，不能用布隆。

常见坑：BITCOUNT 统计口径要和产品对齐——点赞数展示以计数表为准，bitmap 只服务"是否点赞"这一件事，别一物两用。顺带预告：课程第 6 章"用户签到"用的是同一把刀——key=userId、offset=一年里的第几天，GETBIT 扫一圈就能算连续签到，一年 365 bit ≈ 46 字节/用户。

小收口：bitmap = 用"位"存布尔关系，O(1) 判断、极致省内存；记住"1 亿用户 ≈ 12MB"这笔账，面试现场算出来就是加分项。

3.9 实战演进：点赞用户列表的缓存——从慢 SQL 到冷热分离

先抛问题。产品再提需求：“点开点赞列表，要看到是哪些人赞了。“热门内容的点赞用户可能有几百万，怎么存、怎么查、怎么缓存？

V1：JSON 整存整取（最直觉的错误答案）。把点赞用户 id 查全、拼成 JSON、整串塞进一个 String key。三个死穴：构造时要回表查几百万人（慢 SQL）；每次读全量反序列化（CPU 爆炸）；热门内容一个 key 几十 MB——大 key 击穿（06 章 3.26 的诊断在这里全部应验）。

V2：集合存储 + 延长过期。换成 Redis List / Set 存 id 列表，TTL 快到时（低于 1/3）异步续期。击穿缓解了，但大 key 还在——几百万元素的集合照样阻塞单线程。

V2.1：大 key 打散分片。按用户 id 把列表切成固定大小的片：

// 每片 1 万个 uid：like_users:{cid}:slice1 => [uid...], slice2 => [uid...]
slice := (uid % totalSlice) + 1
key := fmt.Sprintf("like_users:%d:slice%d", cid, slice)
rdb.SAdd(ctx, key, uid)
// 分页读取：第 page 页 => 定位到第 page/pageSize 片，LRANGE 范围可控

读哪片查哪片，阻塞和网络放大都解决了——这就是 06 章"大 key 拆分"的落地版。

V3：按用户维度反向组织（优化的精髓）。换个视角：产品真正的高频诉求其实不是"这条内容有谁赞了”（几百万人），而是”我最近赞过什么"（我的主页-点赞 tab）。于是按 userId 建 hash：

key:   user_likes:{uid}           —— 用户维度的点赞记录
field: ttl / cid1..cidN / minCid  —— minCid 是"冷热分界线"
value: 点赞时间等元信息

• 冷热分离：cid > minCid 的是热数据（近期点赞，常驻内存）；更早的算冷数据，翻旧账再回源 DB——内存里永远只留最近一小段；
• 免 TTL 接口调用：过期判断用 hash 里自己的 ttl 字段完成，不用对每个 key 单独调 EXPIRE（那会让接口调用次数翻倍）；
• 一次 HGETALL 到位：“我赞过没 + 我最近赞了啥"一个命令全拿。

参照系是抖音的产品策略：“只看最近 7 天 + 只看铁粉/朋友”——用产品规则把"全量"裁成"最近的一小片”，存储问题就消失了大半。

进阶延伸：课程最后还秀了一段 Redis 源码级的内存优化思路，方向值得记住——用紧凑的自定义结构（定长 int64 id + 位标志复用字段）代替通用容器、用 unsigned short 存小计数（超上限再溢出到字典）、冷数据分层到 SSD 按需加载。Go 里等价的手法是 []byte 紧凑编码 + sync.Pool 对象复用（01 章补充块提过）。面试不用展开实现，讲清"结构对齐数据特征，内存就能压一个数量级“这个思想就够了。

常见坑：V1 → V3 的每一步都由真实痛点驱动（慢 SQL → 大 key → 接口放大），面试讲这个案例切忌跳步直接说"我用 V3”——演进过程本身就是答案。

小收口：至此互动服务的"写、读、判重、列表"四条线全部闭环。3.6~3.9 是课程大纲里计数服务模块的精华映射，配合 3.5 的 Kafka 主线，你已经具备了"高并发互动服务"的完整话术。

四、跟着写：分步实现

本章动手主线是：先把互动服务的存储层建好，再起 Kafka、跑通"发一条消息 → 消费 → 计数 +1"的完整链路。每一步结束时项目都能编译、能跑、能验证。

步骤 1 · 建四张表

目标：把阅读/点赞/收藏的存储结构落库。

代码（在 MySQL 客户端执行，文件 dao/schema.sql）：

-- dao/schema.sql
CREATE TABLE `interactive` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `biz` varchar(32) NOT NULL,
  `biz_id` bigint NOT NULL,
  `read_cnt` bigint NOT NULL DEFAULT 0,
  `like_cnt` bigint NOT NULL DEFAULT 0,
  `collect_cnt` bigint NOT NULL DEFAULT 0,
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `biz_biz_id` (`biz`, `biz_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE `user_like_biz` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `biz` varchar(32) NOT NULL,
  `biz_id` bigint NOT NULL,
  `user_id` bigint NOT NULL,
  `status` tinyint NOT NULL DEFAULT 1,
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `biz_biz_id_user_id` (`biz`, `biz_id`, `user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE `collection` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `name` varchar(128) NOT NULL,
  `user_id` bigint NOT NULL,
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  KEY `idx_user_id` (`user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE `collection_item` (
  `id` bigint UNSIGNED NOT NULL AUTO_INCREMENT,
  `collection_id` bigint NOT NULL,
  `biz` varchar(32) NOT NULL,
  `biz_id` bigint NOT NULL,
  `user_id` bigint NOT NULL,
  `ctime` bigint NOT NULL,
  `utime` bigint NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_collection_biz` (`collection_id`, `biz`, `biz_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

验证：

mysql -u root -p content_community < dao/schema.sql
mysql -u root -p content_community -e "SHOW TABLES;"

预期输出：列出 interactive / user_like_biz / collection / collection_item 四张表。

步骤 2 · 起 Redis，用 Lua 体验原子性

目标：亲手验证"检查 + 自增"的 Lua 脚本。

命令（先有 Redis，可用 docker run -d -p 6379:6379 redis:7）：

redis-cli
# key 不存在时，应返回 0（不操作）
EVAL "if redis.call('EXISTS', KEYS[1]) == 1 then return redis.call('HINCRBY', KEYS[1], ARGV[1], ARGV[2]) else return 0 end" 1 interactive:article:123 read_cnt 1
# 先 HSET 造一个存在的 key
HSET interactive:article:123 read_cnt 10
# 再跑脚本，应返回 11
EVAL "if redis.call('EXISTS', KEYS[1]) == 1 then return redis.call('HINCRBY', KEYS[1], ARGV[1], ARGV[2]) else return 0 end" 1 interactive:article:123 read_cnt 1

预期输出：第一次 0，造数据后第二次 11。这就证明了"检查 + 自增"作为一个整体执行。

步骤 3 · 起 Kafka（Docker）

目标：本地跑起一个 Kafka，监听 localhost:9092。

代码：见 3.5.6 节的 docker-compose.yaml，直接用那一份。

验证：

docker compose up -d
docker compose ps
# 进容器建一个 topic
docker exec -it $(docker ps -qf "name=kafka") /bin/sh -c \
  "cd /opt/bitnami/kafka/bin && kafka-topics.sh --bootstrap-server localhost:9092 --topic read_article --create --partitions 3 --replication-factor 1"

预期输出：Created topic read_article.（若已存在会提示已存在，也正常）。

步骤 4 · 写生产者，发一条 read_article 消息

目标：用 Sarama 同步生产者发一条消息，看到发送成功日志。

代码（文件 mq/producer.go，参考 3.5.13 的 NewSyncProducer 与 Produce）：

// cmd/produce_main.go
func main() {
	p, err := NewSyncProducer([]string{"localhost:9092"})
	if err != nil {
		log.Fatal(err)
	}
	defer p.Close()
	evt := ReadArticleEvent{Uid: 1, Aid: 123}
	payload, _ := json.Marshal(evt)
	// key 用文章 id，保证同一篇文章进同一分区
	if err := Produce(p, "read_article", "123", string(payload)); err != nil {
		log.Fatal(err)
	}
	log.Println("发送成功")
}

验证：

go run ./cmd/produce_main.go

预期输出：发送成功。

步骤 5 · 写消费者组，消费并计数 +1

目标：消费 read_article，调用 IncrReadCnt 把阅读数 +1，验证数据库真的变了。

代码（文件 mq/consumer.go + mq/read_article_handler.go，参考 3.5.13、3.5.9）。消费逻辑里 IncrReadCnt 走的是 3.2 节的 Upsert 落库 + Redis 缓存。

验证：

# 终端 1：起消费者
go run ./cmd/consume_main.go
# 终端 2：发一条消息
go run ./cmd/produce_main.go

预期输出：

• 消费者日志打印处理成功（如 处理阅读事件 aid=123）。
• 查库：mysql -e "SELECT read_cnt FROM interactive WHERE biz='article' AND biz_id=123;" 能看到 read_cnt 从 0 变成 1（或在上次基础上 +1）。

这就跑通了"发消息 → 异步消费 → 计数 +1"的完整链路。

步骤 6 · 故意验证"重复消费"

目标：理解 MarkMessage 的重要性。

做法：把消费逻辑里的 sess.MarkMessage(msg, "") 注释掉，重启消费者，再用步骤 4 发一条消息。

预期现象：因为没提交 offset，消费者重启后会从旧 offset 重新拉，这条消息被再处理一次，数据库 read_cnt 多 +1。验证完记得把 MarkMessage 加回来。

本章完成后的项目结构（互动相关部分）

content-community/
├── dao/
│   ├── schema.sql              # 四张表
│   └── interactive.go           # Upsert / 事务 / 批量
├── cache/
│   └── interactive_redis.go     # Lua 脚本计数
├── biz/
│   └── interactive.go           # InteractiveService：Like/CancelLike/IncrReadCnt
├── domain/
│   └── event.go                 # ReadArticleEvent
├── mq/
│   ├── producer.go              # Sarama 生产者
│   ├── consumer.go              # 消费者组启动
│   └── read_article_handler.go  # 消费逻辑
└── docker-compose.yaml          # Kafka

验证清单（可勾选）

• 四张表已建，联合唯一索引存在
• Redis 里 Lua 脚本"检查 + 自增"返回符合预期
• docker compose ps 显示 Kafka 健康
• read_article topic 已创建，3 个分区
• go run 生产者打印"发送成功"
• 消费者消费后，数据库 read_cnt 真的 +1
• 注释掉 MarkMessage 后会出现重复消费（验证完恢复）

五、工程实践要点

来自原文「工程实践要点」+ 本章经验补充，按优先级排列：

1. 计数类业务优先用 Redis Hash + Lua，不要 SELECT COUNT 查数据库。高频计数走内存，数据库只做最终落库。
2. 自增用 SQL 原子操作（cnt = cnt + 1），不要先读后写——先读后写在并发下会少记（read_cnt 应 12 却记成 11）。
3. Upsert 几乎是必需的：你永远不知道数据库里有没有这条记录，clause.OnConflict 一条语句搞定"有就更新、没有就插入"。
4. 缓存一致性未必都要解决：阅读数、点赞数这类"展示型计数"差几个用户无感；只有余额、库存这类强一致数据才该走事务。
5. Kafka 选 acks 要看业务：金融/订单用 -1，日志/计数可以用 1 甚至 0，别无脑 -1 白白牺牲吞吐。
6. 消费者记得 MarkMessage：不提交 offset 下次还会消费，导致重复处理；想要幂等就把写库逻辑做成 Upsert。
7. 批量接口是性能利器：消费侧攒批 + 生产侧聚合，把"100 次事务提交"变"1 次"，吞吐差一个数量级。
8. 消费者组用 context 控制退出：发版时主动 cancel 优雅退出，别让 Kafka 靠心跳超时判定你死亡，减少无谓 Rebalance。
9. 软删除优于硬删除：点赞/收藏用 status 字段表达取消，性能好且保留行为数据。

六、自测题与动手练习

自测题（合上书能答出来，才算懂）：

1. 阅读、点赞、收藏为什么能抽象成同一个 Interactive 服务？biz + bizId 这个模型解决了什么问题？
2. 阅读计数为什么"先 SELECT 读出现值再 UPDATE“会少记？正确写法靠什么保证并发安全？
3. “检查 key 是否存在 → 存在则自增"为什么必须用 Lua 脚本？Redis 单线程和这有什么关系？
4. 硬删除和软删除在点赞/收藏场景各有什么优劣？为什么实践上偏好软删除？
5. Kafka 里"一个分区在同一时刻只能被同一个消费者组内的一个消费者消费"会导致什么后果？想加快消费速率为什么不能只加消费者？
6. acks=0 / 1 / -1 分别等谁确认？金融交易和阅读计数分别该选哪个，为什么？
7. ISR 是什么？acks=-1 时"不丢消息"的底气来自哪里？
8. 用 Kafka 改造阅读计数后，如果消费者忘记 MarkMessage 会怎样？怎么给"阅读记录"这种新功能零侵入地加能力？

动手练习（建议真做一遍）：

1. 复用互动服务：用 biz="comment"、bizId=456 的思路，给"评论"也接上点赞能力，体会不用新写一套代码的复用（只需换 biz 调用 Like(ctx, "comment", 456, uid)）。
2. 体验 Lua 原子性：在 Redis 里用 EVAL 跑一遍 luaIncrIfPresent 脚本，观察 key 不存在 vs 存在时返回值的差别（应分别是 0 和自增后的值）。
3. 跑通领域事件：用 Sarama 向 read_article 发一条消息，再起一个消费者组消费它，验证计数真的 +1；然后故意在消费里 panic 或注释掉 MarkMessage，看消息会不会被重复投递。

七、模拟面试问答

这一节模拟真实面试场景。每题按问题 → 一句话结论 → 展开 → 面试官想听的关键词 → 常见失分点五段组织。

八、本章小结

学完这一章，下面这些应该能从脑子里调出来。建议逐条打勾，打不上勾的回去看对应小节。

下一章我们将进入可观测性——把"出了问题怎么排查"补齐。你会看到 Kafka 的消费延迟、Rebalance 次数、消息积压量，正是 Prometheus + Grafana 要重点监控的指标；本章埋下的"异步后排查更难"的伏笔，会在下一章用链路追踪和日志来收口。