第18章 · ELK 与 Canal（正式版）

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

这是《内容社区》课程第 18 章，属于第四阶段「高阶业务」的基础设施篇。第 17 章你把搜索服务搭了起来，但当时留了一个尾巴：帖子和 ES 之间的同步靠「双写」，双写失败要靠失败表重试 + 每日全量兜底——全量刷一遍又贵又慢。本章要补上两件生产必备的本领：系统出问题时怎么快速定位（日志怎么收集、怎么查），和数据库变更怎么实时流出去（不用改一行业务代码）。

前者叫 ELK（ElasticSearch + Logstash + Kibana，外加轻量采集器 Filebeat），后者叫 Canal——一个把自己伪装成 MySQL 从库的「抄写员」。你会发现两件事在深处是连着的：Canal 转发变更消息靠 Kafka（第 07 章），Canal 的位点管理思想和 Filebeat 的 registry、Kafka 的 offset 是同一个思想；而 ELK 里那个 E，就是你第 17 章刚学的 ElasticSearch——同一个引擎，日志场景换一种用法。

这一版做了三件事：

1. 每个概念先补课再使用。binlog、主从同步、JVM、ETL、CDC、Grok、位点、GTID……全部就地补「补课胶囊」，不让你去搜。
2. 用类比先让你看见。快递分拣、档案室、抄写员、点外卖查骑手位置——日志和数据同步的世界生活里全有原型。
3. 结尾有 12 道面试问答。按「基础 → 进阶 → 追问」三档，每题告诉你一句话结论、怎么展开、面试官想听什么词、新手最容易答错在哪。

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

一、本章导读

先看一个每个后端都会经历的凌晨现场，感受一下这一章在解决什么问题：

事故现场：凌晨 1 点 47 分，告警群弹出「支付失败率 3%」。你打开电脑要干的第一件事是什么？——看日志。可问题来了：社区现在有 6 个微服务（user / post / comment / relation / search / payment），每个服务 3~5 个实例，散落在十几个容器里。你只能挨个 docker logs、挨个 ssh 上去 grep 那个订单号。等你在第 11 台机器上终于找到那条 callback verify failed 时，四十分钟过去了——而这条日志 30 秒前就产生在第一台机器上。

单机时代 tail -f app.log 一把梭的日子，在多机器、多服务面前彻底结束了。日志还散落在两个更麻烦的维度上：

1. 量：每秒几万条，机器上存三天就把磁盘吃满——你不能全存，又不敢删关键的；
2. 查：MySQL 的数据能用 SQL 查，Redis 的数据能用命令查，散落在几十个文件里的日志用什么查？总不能靠 grep 一台台跑。

本章上半场解决这个问题的答案叫 ELK：让每台机器上的日志自动「流」进一个集中、可检索、可画图的地方。下半场解决另一个问题——第 17 章结尾埋的伏笔：

双写的一致性缝隙，能不能从根上堵住？帖子的每一次变更（不管是哪个服务改的、哪个定时任务改的、甚至 DBA 手敲 SQL 改的），能不能在发生的那一刻就自动同步出去？

答案叫 Canal：它伪装成 MySQL 的从库，让主库把 binlog 主动推给它，解析成消息丢进 Kafka——业务代码一行不改，数据变更自动流出去。这等于给系统装上了「数据库事件总线」：搜索同步、缓存精准失效、审计日志，都能挂在它上面。

学完这一章，你会亲手搭出两条链路：日志链路（服务打一条 error 日志，几秒后在 Kibana 大屏上搜到它）和数据同步链路（UPDATE 一行 post 表，1 秒内 ES 里的文档跟着变，删一行 ES 文档跟着消失）——并且能在面试里把「线上日志体系怎么搭」「Canal 为什么能拿到数据库变更」「为什么按主键 hash 分区」这几道高频题答得干干净净。

开读前先看一眼全局：本章是两条互不相干的流水线——日志链路（服务打日志 → Filebeat → Kafka → Logstash → ES → Kibana）和数据同步链路（MySQL binlog → Canal → Kafka → 同步 ES）。把这两条线分开记，正文的每个组件你都知道它站在哪一节管道上。

flowchart LR
    subgraph ELK["日志链路：一条 error 日志的旅程"]
        A["服务打一条日志"] --> B["Filebeat 采集"]
        B --> C[("Kafka")]
        C --> D["Logstash 解析"]
        D --> E[("Elasticsearch")]
        E --> F["Kibana 检索看板"]
    end
    subgraph CANAL["同步链路：一次 UPDATE 的旅程"]
        G["业务 UPDATE MySQL"] --> H["binlog"]
        H --> I["Canal 伪装从库
拉取变更"]
        I --> J["投递 Kafka"]
        J --> K["消费者同步 ES"]
    end

这张图怎么读：上下两个子图是本章要搭的两条互不相干的流水线。上面日志链路：服务打一条日志，由部署在每台机器上的 Filebeat 采集，先丢进 Kafka 做缓冲，Logstash 消费并解析成结构化字段后写入 Elasticsearch，最后统一在 Kibana 检索看板里查。下面同步链路：业务 UPDATE 落进 MySQL 产生 binlog，伪装成从库的 Canal 拉取并解析变更、投递 Kafka，消费者再把它同步进 ES。两条链路都借 Kafka 削峰解耦，但彼此独立——日志搜不到就查上半条，ES 不同步就查下半条，分段排障是本章的动手主线。

二、本章知识点大纲

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

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

1. 用 docker compose 起一套 Filebeat + Logstash + ES + Kibana（ES 复用第 17 章的），把自己 Go 服务的 JSON 日志收进 Kibana，能按 data.level、data.trace_id、data.duration 检索。
2. 写出一条 Logstash 管道（input / filter / output）：json 解析、date 时间标准化、drop 噪声、按天分索引输出——并解释每个 filter 为什么在那里。
3. 部署 Canal：MySQL 开 ROW 模式 binlog、建最小权限的 canal 账号、配置转发 Kafka，用 kafka-console-consumer 亲眼看到一条 binlog 消息。
4. 写 canal-consumer：消费 binlog 消息按 database + table 路由，post 表变更同步 ES；幂等（文档 _id = 主键、DELETE 遇 404 当成功），失败消息转死信 topic。
5. 给日志索引配按天分索引 + ILM 保留策略（本地学习环境至少做到按天分 + 定期清理）。

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

1. ELK 的「采集 → 处理 → 存储 → 展示」四段式，以及为什么生产上是 Filebeat 蹲机器、Logstash 集中处理的「轻采集、重处理」分工。
2. Logstash 三段式与四大 Filter 插件（Grok / Date / Mutate / Kv）各自干什么；为什么复杂逻辑应该在日志源头规范掉而不是堆在 Logstash。
3. Canal 的本质：伪装从库走 MySQL 复制协议（注册 slave → dump 请求 → 主库推送 → 解析投递），以及位点（binlog 文件名 + offset）怎么支撑断点续传。
4. binlog 三种模式（Row / Statement / Mixed）的取舍，为什么 Canal 场景几乎都用 Row（NOW() / UUID() 重放不一致的坑）。
5. 顺序性：为什么同一主键的消息必须 hash 到同一分区，否则「旧值覆盖新值」；以及 Canal 位点只记「投递成功」不记「消费成功」→ 下游必须幂等。
6. Canal 消息格式：data / old / type 三字段——DELETE 的被删数据在 data 不在 old；列值全是字符串。
7. 增量同步两个案例的取舍：直接用 binlog 数据回写缓存 vs 回查 DB；数据校验「以谁为准」的切换陷阱。
8. 可观测性全景：日志（ELK）、指标（Prometheus）、链路（OTel）三支柱与 Kibana / Grafana 的分工。

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

1. ILM 四阶段（Hot / Warm / Cold / Delete）与 ES 磁盘水位线（85% 迁移分片、95% 索引只读）的坑。
2. Filebeat 的 registry 文件：断点续传的机制，以及「删 registry 日志翻倍」「轮转过快日志丢失」两个坑。
3. GTID（全局事务标识）：主从切换后 Canal 位点失效时的高可用思路（面试加分点）。
4. ZooKeeper 在 Canal 集群里的角色（位点共享、HA 协调）。
5. Canal Client 拉模式 vs Kafka 投递模式的区别（本课程用后者）。

前置依赖与补课清单

不需要你提前懂的：Lucene 段合并细节、Logstash 全部插件手册（用到哪查哪）、Canal 集群部署与 admin 控制台（本地单实例够用）、Kibana 可视化图表的全部玩法（Discover 会用即可）。

预计时长与学习成果

• 预计时长：5 小时（概念精读 2 小时 + 跟着搭两条链路 3 小时）；本章概念较多（日志链路、伪装从库、增量同步三个新核心），建议分两次学：第一次读完 3.1–3.6 的日志侧，第二次学 3.7–3.9 的 Canal 侧并做「跟着写」。

• 学完你能跑出来的东西（这是本章的验收标准，做到才算过关）：

  两条链路全通。日志侧：调用一次登录失败接口，几秒后在 Kibana（localhost:5601）的 Discover 里搜到这条 error，能按 data.level、data.trace_id 过滤。数据侧：UPDATE post SET title='端到端验证' WHERE id=1 执行后 1 秒内，curl localhost:9200/post/_doc/1 返回新标题；DELETE 该行后同一请求返回 404；把 ES 停掉再改库，失败消息躺在 canal-events-dlq 死信 topic 里，恢复后重放，ES 结果不变（幂等验证）。

本章的学习路线（先日志侧后数据侧，最后跟着写把两条链路都跑通）：

flowchart LR
    A["3.1-3.3 日志链路
ELK·Logstash 管道"] --> B["3.4-3.6 Filebeat 分工
Kibana·ILM 治理"]
    B --> C["3.7-3.8 Canal 原理
伪装从库·binlog"]
    C --> D["3.9 增量同步
缓存·校验·保序"]
    D --> E["第四节 跟着写
两条链路落地"]
    E --> F["第七节 面试问答"]

这张图怎么读：从左到右的单向路线，先日志侧后数据侧。3.1-3.3 先把 ELK 四段式和 Logstash 管道讲透，3.4-3.6 补 Filebeat 分工、Kibana 用法与按天分索引、ILM 日志治理，3.7-3.8 切到 Canal（伪装从库的复制协议四步、binlog 三模式与消息格式），3.9 用缓存更新、数据校验两个案例把变更流真正用起来，最后进第四节跟着写把两条链路都落地，以第七节面试问答收尾。

三、正文

3.1 单机日志的绝境：为什么需要一条「日志流水线」（核心概念一）

先抛问题。回到开头的凌晨现场：6 个服务 × 每个几个实例 = 几十个日志文件散落在十几台机器上。你要回答三个最基本的问题：「这笔订单的完整请求链路经过了哪些服务」「过去 10 分钟 error 每分钟爆发多少条」「上个周四下午那个诡异超时还有日志留存吗」——tail -f 和 grep 一个都答不了。不解决会怎样？排障时间以小时计，且全靠运气：日志在哪台机器、还在不在，都是玄学。

给直觉。把这件事类比成快递网络：每家每户（每台机器上的每个服务）都在往外寄信（打日志）。你不能让收信人挨家挨户上门翻信箱（SSH 逐台 grep）——正确的做法是：每家门口放一个邮筒（Filebeat，轻量采集），信件统一寄到分拣中心（Logstash，集中处理：拆信、贴标签、归类），归档进可检索的档案室（ES 存储 + 建索引），前台再给一个查档窗口（Kibana 查询与画图）。信一旦进了档案室，你就能按寄件人、按日期、按关键词秒级查到任何一封信。

这条「家门口 → 分拣 → 归档 → 查档」的路，就是本章的第一个核心概念——日志链路：

应用日志 → Filebeat（采集） → Logstash（处理） → ES（存储） → Kibana（展示）

最小可运行示例。先感受一下「日志链路」要取代的原始生活——单机看日志你早就会了，但给你 30 秒体会一下它的极限在哪：

# 单机时代：跟着日志滚动看（-f = follow，有新日志就打印）
tail -f logs/payment-service.log

# 多机时代你要干的活：挨台机器搜同一个订单号（这就是要被消灭的工作方式）
ssh deploy@10.0.0.11 "grep 'order_1024' /var/log/payment/*.log"
ssh deploy@10.0.0.12 "grep 'order_1024' /var/log/payment/*.log"
# ……还有十几台。而且你不知道昨天的日志被轮转压缩成 .gz 了吗？

逐行讲解：tail -f 只能盯一个文件；grep 只能搜一台机器。这两个命令没有跨机器、跨时间、按字段过滤的能力——注意「按字段」三个字：你的日志明明是结构化 JSON（第 08 章用 zap 打的），grep 却只能把它当纯文本糊搜。日志明明有结构，却用无结构的方式查，这是第二重浪费。

常见坑：新手会说「我把所有服务的日志都打到一台机器上不就行了」——用 NFS 共享目录或者 scp 汇总。这在日均几 MB 的 toy 项目里能跑，但生产环境多实例并发写同一个文件会互相踩踏（日志交错、丢失），而且没有解析、没有索引，几十 GB 的文本文件 grep 一次照样几分钟。集中 ≠ 可检索。

小收口：这一节确立的目标是搭一条日志流水线，让日志自动「流」向集中可检索的地方。流水线上有四个工位，下一节挨个认识它们。

3.2 ELK 三件套：三个组件各管一段

先抛问题。要建「分拣中心 + 档案室 + 查档窗口」，为什么偏偏是 E、L、K 这三个组件？它们各自解决什么？——不知道分工，出了问题都不知道去查哪个组件的日志。

给直觉。还记得第 17 章的 ES 吗？它是个「为检索而生」的分布式存储。日志的本质恰好是海量文本 + 按词检索——这不就是 ES 的主场吗？所以存储和检索这个工位，直接复用你已经学过的 ES，一件新东西都不用学。真正要新认识的是 L（分拣）和 K（查档窗口）：

（后来实践中又加了一个 Filebeat 负责「门口邮筒」的角色，所以严格说叫 EFK / ELK+Filebeat，但大家都习惯还叫 ELK。）

最小可运行示例。先用 30 秒体会「日志进 ES 之后的世界」——第 17 章你已经在 ES 里存过帖子文档了，日志和帖子在 ES 眼里没有任何区别，都是文档：

# 第 17 章查帖子文档用过的命令，同样适用于查日志文档
curl -s "localhost:9200/post/_search?q=并发&pretty"
# 将来日志进 ES 后，Kibana 干的事本质上就是这个（只是给了你一个点击界面）：
#   查 data.level 为 error 的文档、按时间排序、聚合每分钟多少条……

逐行讲解：这条命令揭示了一个关键事实——ELK 里没有任何新魔法。日志一旦变成 ES 里的文档，你第 17 章学的 term / match / bool / aggs 全部原样适用：「按级别过滤」就是 term 查 keyword 字段（filter 篮子），「每分钟 error 趋势」就是按时间聚合。Kibana 只是把这些查询变成了点点点的界面。

常见坑：新手容易把 ELK 当成「一个软件」装。它是三个独立进程，各有各的端口（ES 9200、Logstash 5044/9600、Kibana 5601），各有各的配置文件——链路哪一环断了，日志就停在哪一环，排障时要分段定位（第 4 步验证时会教你逐段排查）。

小收口：E 复用第 17 章，K 后面专门讲，剩下最大的新东西是 L——Logstash 这根「分拣管道」内部长什么样，下一节拆开看。

3.3 Logstash：一根 Input → Filter → Output 的管道

先抛问题。日志从 Filebeat 原封不动送过来时是「一行 JSON 字符串」，凭什么 Kibana 里就能按 level、service 过滤、按时间排序？中间必须有人把这行字符串拆开、转成结构化字段、修正时间戳、扔掉垃圾字段——这个「人」就是 Logstash。

补课胶囊 · ETL：ETL 是 Extract-Transform-Load 的缩写，「抽取 → 转换 → 加载」的数据搬运三段式 → 类比机场行李分拣：从传送带取行李（Extract）、按目的地贴标签分拣（Transform）、装上对应航班（Load）→ Logstash 就是一个专门搬日志的 ETL 工具 → 回到主线：Logstash 把这三段做成了声明式配置。

给直觉。Logstash 的世界观极其简单，就三个房间：

flowchart LR
    I["Input 输入
从哪收（文件/端口/Kafka）"] --> F["Filter 过滤
解析·转换·清洗"]
    F --> O["Output 输出
发到哪（ES/文件/Kafka）"]

数据从左进右出，中间的 Filter 房间里可以串任意多个插件，一个接一个地加工数据。配置文件就是这个房间的声明。

最小可运行示例。下面是一条最简管道：从文件读日志，用 Grok 解析，写进 ES：

# logstash.conf —— Logstash 管道配置（三个大括号对应三个房间）
input {
  file {
    path => "/var/log/webook/*.log"   # 监听哪些文件
    start_position => "beginning"     # 首次启动从文件开头读（默认只读新增）
  }
}

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:msg}" }
  }
}

output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "webook-logs-%{+YYYY.MM.dd}"  # 按天分索引（3.6 节讲为什么）
  }
}

逐行讲解：input.file 声明「从这些路径收日志」；filter.grok 用一行「模板 + 捕获」把非结构化文本切出三个字段（下面马上拆）；output.elasticsearch 声明写进哪个 ES、索引名按日期取值——%{+YYYY.MM.dd} 表示「取事件时间、格式化成年.月.日拼进索引名」，于是每天自动生成一个新索引。注意这份配置不用背，用到时查文档或问 AI 都行，你要记住的是三个房间的思维模型。

Filter 房间里的四个常用插件

（1）Grok：把一行文本切成字段。本质是「带名字的正则捕获」：

补课胶囊 · Grok：Grok 是 Logstash 的正则模板库，「给常用正则起好名字」的语法糖 → 类比切蛋糕的模具：%{TIMESTAMP_ISO8601:timestamp} 的意思是「用 ISO 时间这个现成模具，切下来的那块叫 timestamp」→ 最小示例见上面管道里的 match 行 → 回到主线。

内置模板覆盖了常见格式（TIMESTAMP_ISO8601、IP、EMAIL、LOGLEVEL、GREEDYDATA 贪婪匹配剩余全部），不用自己写正则。它主要服务于「非结构化」的老式文本日志——而你的 zap 日志本来就是 JSON 单行，所以本章「跟着写」里用的是 json 插件而不是 Grok（这是很多教程没说清的：Go 结构化日志 + Grok 是脱裤子放屁，JSON 直接解析更快更稳）。

（2）Date：把字符串时间变成 ES 认的 @timestamp。日志里自带的时间是字符串，ES 需要真正的 date 类型字段才能按时间排序、做时间范围查询：

filter {
  date {
    match => ["[data][ts]", "ISO8601"]   # 从 data.ts 字段读时间，按 ISO8601 解析
    target => "@timestamp"               # 写进 @timestamp（ES 默认时间字段）
  }
}

为什么要多此一举？因为默认的 @timestamp 是 Logstash 收到这条日志的时间——如果 Filebeat 重启、积压了一批半小时前的日志，时间就全漂了。用日志产生时的时间覆盖它，Kibana 里的时间轴才是真的。

（3）Mutate：对字段做增删改。重命名、删字段、加字段、类型转换：

filter {
  mutate {
    add_field => { "env" => "dev" }        # 添加字段
    remove_field => ["message", "agent"]   # 删字段（原始报文、Filebeat 元数据）
    rename => { "old_name" => "new_name" } # 重命名
  }
}

（4）Kv：从文本里抠键值对。按分隔符提取 k=v 对（比如 a=1,b=2 的半结构化文本）。对 Go 结构化日志基本用不上，见到认识即可。

常见坑：① 配置文件语法错误（少了逗号、括号不配对）时 Logstash 启动直接失败，docker logs logstash 里能看到具体行号——看到 Pipeline started 才算配置通过；② Grok 正则写错不会报错，只会打上 _grokparsefailure 标签让日志「消失在解析失败的角落」，Kibana 里搜这个标签能找到受害者。

小收口：分拣中心看完了。但你可能注意到一个疑点：既然 Logstash 也有 input.file 能直接读文件，为什么图里还要在它前面加一个 Filebeat？这正是下一节的问题。

3.4 Filebeat：为什么「采集」和「处理」要分工

先抛问题。Logstash 完全可以自己蹲在每台机器上读日志（上一节的 input.file 就是证据），为什么生产上还要多加一个组件？——因为 Logstash 有一个致命的体质问题。

补课胶囊 · JVM：JVM 是 Java Virtual Machine（Java 虚拟机）的缩写，Java 程序跑在它上面 → 类比「带着整个厨房搬家」：Go 编译出的二进制像泡面（开水一冲就能跑，占资源极小），Java 程序像开餐厅（先搬灶台、冰箱、水电——JVM 启动就要几百 MB 内存，跑起来还要 GC 垃圾回收吃 CPU）→ Logstash 是 Java 写的，就是个「自带厨房的大厨」→ 回到主线：大厨手艺好（Filter 什么都能干），但你不会在每家每户都雇一个大厨。

给直觉。分工的原则：轻采集、重处理。每台业务机器上只放一个 Go 写的轻量邮差（Filebeat，内存占用几 MB，只干一件事：盯住日志文件、把新增的行发走）；处理逻辑（解析、清洗）集中部署在一小组 Logstash 上。这样业务机器不被拖累，处理能力又可以按需扩容：

flowchart LR
    subgraph M1[业务机器 1]
        A1[应用日志] --> F1[Filebeat 轻采集]
    end
    subgraph M2[业务机器 2]
        A2[应用日志] --> F2[Filebeat 轻采集]
    end
    subgraph M3[业务机器 3]
        A3[应用日志] --> F3[Filebeat 轻采集]
    end
    F1 --> LS[Logstash 集中处理]
    F2 --> LS
    F3 --> LS
    LS --> ES[(ES 存储)]
    ES --> KB[Kibana 展示]
    F1 -. 也可跳过 Logstash 直写 .-> ES

Filebeat 的完整优势清单：轻量高效（Go 编写）、实时监测文件变化、模块化支持常见格式（NGINX / 系统日志等）、多目的地输出（Logstash / ES / Kafka）、自动发现新日志文件、容器环境友好。虚线那一条「直写 ES」要注意：小场景确实可以跳过 Logstash 直写，但那就失去了集中处理能力——本章走标准链路。

最小可运行示例。Filebeat 的配置就两个问题：收哪些文件、发给谁：

# filebeat.yml
filebeat.inputs:
  - type: filestream
    id: content-community          # 数据源 ID，registry 靠它识别
    enabled: true
    paths:
      - /var/log/content-community/*.log   # 要采集的日志文件（容器内路径）
    fields:
      env: dev                     # 附加环境标签，Kibana 里按环境过滤
    fields_under_root: true

output.logstash:
  hosts: ["logstash:5044"]         # 发给 Logstash，不直写 ES

逐行讲解：filestream 是 8.x 推荐的输入类型（旧的 log 类型已废弃）；paths 支持通配符；fields 给每条日志附加自定义字段——将来 10 个服务的日志混在一起，靠 service 字段就能只看某一个服务。「跟着写」第 2 步会把它挂进 docker compose。

Filebeat 的位点管理（registry）：断点续传的秘密

先抛问题：Filebeat 重启后，为什么不会把日志文件从头再发一遍？也不会从最新处开始、漏掉停机期间的日志？

给直觉：你在读一本厚书时会夹书签——读到哪一页，签就夹到哪一页。Filebeat 的书签叫 registry：本地数据目录下的一个文件，为每个被采集的文件记录「文件路径 + 已读到的偏移量（offset，即读到了文件的第几个字节）+ 文件标识（inode）」。每发成功一批就更新书签；重启后先翻书签，从上次的位置接着读。

flowchart LR
    F[Filebeat] -->|"读文件·记录 offset"| R[(registry 文件)]
    F -->|发送日志| L[Logstash / ES]
    R -->|"重启后从 offset 续读"| F

先别急着觉得眼熟——它确实眼熟：这个思想和第 07 章 Kafka 消费者的 offset、本章后面 Canal 的位点，是同一个思想在不同系统里的化身：记住「处理到哪了」，才能断点续传。面试里能把这三处串成一句话，是很漂亮的回答。

常见坑：① 删了 registry（或容器里没挂载数据目录）：Filebeat「失忆」，整个文件从头再发一遍，Kibana 里日志全部翻倍——所以 registry 目录必须持久化（compose 里挂 volume）；② 日志轮转快于采集：lumberjack 切片 + 压缩删除如果发生在 Filebeat 读完之前，没读完的部分直接丢了。排错口诀：发现日志重复或缺失，第一时间查 registry 的 offset。

小收口：采集（Filebeat）和处理（Logstash）的分工说清了。数据马上要进 ES 了——但进之前还有最后一站要规划：这些日志要建什么样的索引、留多久。这决定你半年后会不会半夜被告警叫醒。

3.5 Kibana：给日志装上「大屏」

先抛问题。日志进了 ES，你完全可以用 curl + 查询 DSL 查（第 17 章学过）——但总不能让值班的同事手写 JSON 查询吧？需要一个人人可用的查询界面和仪表板。

给直觉。Kibana 就是档案室的前台窗口 + 墙上的大屏：窗口让你点点鼠标就能搜（Discover 页）、大屏把数据画成图（Dashboard 页）。它和 ES 是绑定的关系——Kibana 只管「展示 ES 里的数据」，本身不存任何数据。

三个核心功能：

• Discover（发现）：逐条浏览日志、按字段过滤、全文搜索——排障时 90% 的时间都在这一页；
• Dashboard（仪表板）：把多个图表（error 趋势折线图、慢请求分布柱状图、接口状态码饼图）拼成一屏；
• 可视化：柱状图、折线图、地图等多种图表类型。

讲义对 Kibana 的定位原话很生动：侧重数据展示，「主打四个字——花里胡哨」。别被它花哨的图表迷住，排障的核心就 Discover 一页。

Kibana vs Grafana：别选错工具

第 08 章你用过 Grafana（配 Prometheus 看指标），现在又来了个 Kibana——长得很像，怎么分工？

一句话分工：日志查询用 Kibana，指标监控和告警用 Grafana，两者在生产上是共存的黄金搭档。

可观测性全景：三支柱归位

把第 08 章的三支柱和本章的工具对上号，你会得到一张完整的可观测性地图——这是面试「你们公司怎么做可观测性」的标准答案骨架：

flowchart LR
    APP[你的应用] -->|日志| ELK["ELK（本章）
发生了什么"]
    APP -->|指标| PROM["Prometheus（第 08 章）
趋势正不正常"]
    APP -->|链路| OTEL["OpenTelemetry（第 08 章）
慢在哪个环节"]
    ELK --> G[Grafana / Kibana 统一看板]
    PROM --> G
    OTEL --> G

三个维度回答三个不同的问题：日志告诉你「具体发生了什么」（某条请求为什么失败）；指标告诉你「整体趋势正不正常」（错误率突增了吗）；链路告诉你「一次请求慢在哪个环节」（跨服务的耗时分布）。出问题时三者的使用顺序通常是：指标发现异常 → 链路定位环节 → 日志查明细节。

常见坑：Kibana 首次使用必须在 Stack Management → Index Patterns 里手动创建索引模式（如 content-community-logs-*，时间字段选 @timestamp），否则 Discover 页是空的——新手最容易在这一步误以为「链路没通」。「跟着写」第 4 步会专门验证。

小收口：展示端就绪。还剩最后一个问题——日志天天往 ES 里灌，磁盘是有限的。下一节讲日志的「退休制度」。

3.6 日志治理：按天分索引与 ILM 生命周期

先抛问题。按 3.3 节的配置，日志索引是 xxx-logs-%{+YYYY.MM.dd} 按天分的。为什么多此一举按天分，而不是全部塞进一个大索引？——因为删旧日志的方式完全不同。

给直觉。两个仓库整理方式：A 仓库所有货堆成一整垛（一个索引），到期要扔 30 天前的货只能从垛里一件件挑出来扔（DELETE FROM ... WHERE time < ? 的 delete-by-query，慢几个数量级还伤索引）；B 仓库每天的货装一个独立箱子（按天分索引），到期直接整箱搬走扔掉（删一个索引，毫秒级元数据操作）。ES 的索引删除就是这么便宜。

按天分索引还白送两个好处：冷热天然分离（新数据写当天「热」索引、查询也集中在最近几天，旧索引不被打扰）和时间边界天然清晰（2026.08.25 的索引里不可能有 26 号的日志）。

ILM：让索引自动「退休」

只按天分还不够——谁去执行「删除 30 天前的索引」？靠人肉定时任务？ES 内置了索引生命周期管理（ILM，Index Lifecycle Management），把索引的一生划成四个阶段，每个阶段配置对应动作：

ILM 策略的声明长这样（一条「保留 30 天」的最简策略）：

// ILM 策略：日志索引保留 30 天后整体删除
PUT /_ilm/policy/webook-logs-policy
{
  "policy": {
    "phases": {
      "hot": { "actions": { "set_priority": { "priority": 100 } } },
      "delete": { "min_age": "30d", "actions": { "delete": {} } }
    }
  }
}

把策略绑定到索引模板（index template，匹配 webook-logs-* 的索引自动套用这套配置），之后每天新生成的索引就自动继承策略，到期自动删除，全程无人值守。

常见坑（新手必踩，血泪教训）：只建索引不设保留策略。日志每天几个 GB 地涨，某天磁盘水位超过 cluster.routing.allocation.disk.watermark.low（默认 85%），ES 开始疯狂迁移分片自救；超过 95% 则直接把索引置为只读——写入全面报错，整条日志链路瘫痪，半夜被告警叫起来手动删索引的，多半是没配 ILM 的。哪怕本地学习环境，也至少做到按天分索引 + 定期手动清理。

小收口：日志侧（上半场）到此完整了：Filebeat 采集 → Logstash 处理 → ES 存储（按天分索引 + ILM 退休）→ Kibana 展示。下半场换频道——不解决「看」的问题了，解决「数据变更怎么自动流出去」的问题。主角 Canal 登场，而它的一切都建立在 MySQL 的 binlog 之上。

3.7 Canal：伪装成 MySQL 从库的「抄写员」（核心概念二）

先抛问题。第 17 章结尾留的悬念：帖子数据要同步进 ES，当时用的是业务代码双写——发帖接口里调一次 gRPC 把帖子塞给搜索服务。这个方案有三个绕不开的痛：

1. 侵入：每个改帖子的地方（编辑、审核、后台管理、未来的定时任务）都得记得调一次同步，漏一处就是数据不一致；
2. 不是所有变更都经过你的代码：DBA 手敲 SQL 修数据、数据订正脚本直连 DB——这些变更你的代码根本不知道；
3. 双写失败要自己兜底：失败表、重试任务、每日全量——三板斧样样要维护，全量同步又贵又慢。

理想方案是什么？让变更从数据库自己「流」出来，不管是谁改的、怎么改的，只要落进了 MySQL，就能被捕获。Canal 就是干这个的。

补课胶囊 · binlog：binlog（binary log，二进制日志）是 MySQL 自带的变更流水账，记录每一次写入操作（INSERT / UPDATE / DELETE）的详细内容 → 类比银行的流水单：每一笔存款取款都记一笔，且记完就不可篡改——它本来是 MySQL 用来崩溃恢复和主从同步的，但任何人拿到流水单，都能重放出一模一样的账 → 第 07 章提过一嘴，本章正式启用它。

补课胶囊 · MySQL 主从同步：主库（master）把每次变更写进 binlog；从库（slave）主动连上主库请求同步，流程四步：

1. 从库连上主库
2. 从库发起数据同步请求（"把 binlog 从某某位置开始发给我"）
3. 主库开一个线程，把 binlog 持续推送给从库
4. 从库收到后先存到本地的 Relay log（中转日志），再逐步重放执行

→ 类比总店与分店：总店每出一道新菜谱（binlog），分店派人来抄（同步请求），总店安排专人持续递菜谱（dump 线程），分店抄到小本子上（Relay log）照着做 → 回到主线：Canal 要做的事，就是把自己扮演成从库。

补课胶囊 · CDC：CDC 是 Change Data Capture（变更数据捕获）的缩写，泛指「捕获数据库的每一次变更并对外提供」的技术 → 类比「上市公司公告」：数据一变（重大事件），市场（下游系统）立刻知道 → Canal 是 CDC 在 MySQL 生态最流行的开源实现（同类还有 Debezium）。

给直觉。Canal 的白话类比：坐在数据库旁边的抄写员。数据库每改一笔账（INSERT / UPDATE / DELETE），抄写员立刻把这笔变动抄成一张小纸条，送到你指定的地方（Kafka）。你不用改任何业务代码，就能感知到数据库发生的所有变化。它的「应用场景」包括：数据仓库同步、实时分析、缓存更新、数据校验——以及本章要做的搜索索引同步。

Canal 怎么「骗过」MySQL：复制协议四步

「伪装成从库」不是修辞，是字面意义上的照着真从库的行为演一遍。MySQL 主从之间用的是一套复制协议（replication protocol），Canal 严格按剧本走：

1. 注册为 slave：Canal 向 MySQL 发送 COM_REGISTER_SLAVE 命令，报上一个「假 server_id」——每个从库要有集群内唯一的编号，所以 Canal 这个 ID 不能和真实从库冲突；
2. 发起 dump 请求：发送 COM_BINLOG_DUMP，告诉主库「请把从某个 binlog 文件、某个偏移量开始的增量数据推给我」；
3. 主库开 dump 线程推送：在主库眼里，Canal 和一个普通从库没有任何区别，它会持续把 binlog 事件流推过来；
4. Canal 解析并投递：收到二进制的 binlog 事件后，解析成结构化消息（insert / update / delete + 完整行数据），再投递给客户端或 Kafka。

sequenceDiagram
    participant C as Canal
    participant M as MySQL 主库
    participant K as Kafka
    C->>M: COM_REGISTER_SLAVE(server_id=1234)
    C->>M: COM_BINLOG_DUMP(从 mysql-bin.000003 的 offset 4567 开始)
    M-->>C: 持续推送 Binlog 事件流
    C->>C: 解析 Binlog 为结构化消息
    C->>K: 投递 Binlog 消息

整条链路下来，MySQL 以为自己在给一个从库同步数据，实际上数据流进了 Kafka。业务代码零改动——这就是 Canal 解决双写问题的根本姿势。

Go 生态的替代路线：如果不想部署独立的 Canal Server，Go 还有一条轻量路径——用纯 Go 的 go-mysql-org/go-mysql/canal 库，在应用进程内伪装 MySQL 从库直接解析 binlog。省一个组件，但位点管理、HA、投递 MQ 都要自己扛；本课程选独立 Canal + Kafka 转发，是把「数据同步」这个通用能力和业务服务解耦，多个下游服务共享同一份变更流时优势明显。

Canal 的基本组成

位点管理：Canal 的「书签」

又是书签。Canal 消费 binlog 到哪儿了，靠**位点（position）**记录：binlog 文件名 + 文件内偏移量（offset）。Canal 把位点持久化在本地（instance 目录下的 meta.dat 文件）；部署 HA 集群时则存到 ZooKeeper 里（上面组件表里 ZK 的用途之一）。Canal 重启后读取位点，从上一次的位置继续拉取。

常见坑：Canal 连不上 MySQL，九成是 MySQL 侧的问题——binlog 没开 ROW 模式（下一节讲为什么必须是 ROW）、canal 账号缺 REPLICATION SLAVE 权限被 dump 拒绝、或 canal 容器起太早 MySQL 还没就绪。「跟着写」第 5 步会给一张逐项排查清单。

小收口：Canal 靠伪装从库拿到 binlog，把「数据库变更」变成了「Kafka 消息」。但消息长什么样？为什么监听 binlog 前必须把 MySQL 切到 Row 模式？下一节拆开消息看。

3.8 binlog 的三种模式与 Canal 消息格式

先抛问题。binlog 记录「每次变更」，但「怎么记」有三种姿势——姿势选错，下游拿到的消息就是废纸。

给直觉。还是记流水账，两种记法：

为什么 Canal 场景几乎都用 Row？看 Statement 的死穴：主库执行 UPDATE orders SET pay_time = NOW() WHERE id = 1，binlog 里记的是这条语句；从库（或 Canal 的消费者）重放时再执行一次 NOW()——此时时间已经过了几秒，重放结果和主库不一致。而 Row 模式记的是「pay_time 最终等于 2026-09-06 14:30:05」这个真实值，谁重放都是这个值。讲义的判断是「基于行的日志记录用起来比较方便」——对 CDC 来说，Row 是唯一正确答案。

这也是为什么 Canal 的 docker compose 里 MySQL 必须带这两行参数：

# docker-compose.yml（MySQL 服务，关键两行参数）
services:
  mysql:
    image: mysql:8.0
    command:
      - --binlog-format=ROW             # 关键：binlog 用 Row 格式
      - --binlog-row-image=FULL          # 关键：行镜像记完整行（默认也是 FULL）

Canal 消息长什么样：三种变更三个样

INSERT 消息——data 里是插入的完整行，列名是键、列值是字符串：

{
  "data": [
    { "id": "1", "name": "大明", "email": "john@example.com" }
  ],
  "database": "webook",
  "table": "users",
  "type": "INSERT",
  "ts": 1234567890
}

UPDATE 消息——比 INSERT 多了 old 字段，放更新前的旧值。因为上面配了 binlog-row-image=FULL，old 里是完整旧行（所有列都记，不只是被改的列）：

{
  "data": [ { "id": "1", "name": "大明2", "email": "john@example.com" } ],
  "old":  [ { "id": "1", "name": "大明", "email": "john@example.com" } ],
  "type": "UPDATE"
}

DELETE 消息——注意，反直觉的地方来了：被删除的完整数据放在 data 里，old 反而是空的：

data 字段：被删除的数据（完整行，含主键）
old 字段：没有（空）
type：DELETE

小收口：现在你认识消息了。最后一步：把这些消息用起来。两个实战案例——更新缓存和数据校验，后者还是第 11 章数据迁移的「增量校验」环节的升级版。

3.9 增量同步的两个实战案例（核心概念三）

先抛问题。拿到 binlog 消息流之后，怎么把它变成生产能力？讲两个最有代表性的场景：缓存自动更新（解第 06 章留下的双写一致性难题）和数据校验（解第 11 章迁移时的增量核对难题）。

案例一：借助 Canal 更新缓存

第 06 章学缓存时你被「双写一致性」折磨过：更新 DB 和更新缓存是两个动作，中间任何时刻都可能被打断——先删缓存再更新 DB，并发读会把旧值重新灌进缓存；先更新 DB 再删缓存，删除失败缓存就是脏的。Canal 给了一个釜底抽薪的方案：业务代码只管更新 DB，缓存由 binlog 消费者异步维护——DB 一变，消息一来，缓存跟着变。

具体做法有两个流派：

顺序性：方案一的命门。方案一直接用消息里的值覆盖缓存，那就要求消息按变更发生的顺序到达消费者。Kafka 只保证分区内有序——如果 id=1 的 INSERT 和 UPDATE 被发到不同分区，消费者可能先处理 UPDATE（缓存 = 2）再处理 INSERT（缓存 = 1），旧值覆盖新值，缓存从此是错的。

解法你在第 07 章就学过：按 key 分区。Canal 支持按主键 hash 分区，把同一主键的消息钉死在同一个分区：

# canal 的 Kafka 转发配置：按主键 hash 分区
canal.mq.partitionHash = .*\\..*:$pk$

效果：

分区 0: 消息1(id=1, INSERT) → 消息2(id=1, UPDATE a=2) → 消息3(id=1, UPDATE a=3)
分区 1: 消息4(id=2, INSERT)

flowchart LR
    subgraph P0["分区 0：id=1 的消息全在这"]
        M1["INSERT a=1"] --> M2["UPDATE a=2"] --> M3["UPDATE a=3"]
    end
    subgraph P1["分区 1：id=2 的消息全在这"]
        M4["INSERT id=2"]
    end
    P0 --> C["消费者按序回写缓存"]
    P1 --> C

表与 topic 的关系（规模化问题）：小规模时所有表共用一个 topic（本课程做法，canal-events 一个 topic 装所有表的变更，消费者按 table 字段路由）；大规模时不同表拆不同 topic、甚至不同的 Kafka 集群——避免一张大表的变更洪峰把别的表的消息淹没（消息积压 + 集群瓶颈）。

代码实现——一个监听 binlog 更新用户缓存的消费者：

// internal/consumer/binlog_cache.go
type CanalBinlogConsumer struct {
    cache  cache.UserCache
    repo   repository.UserRepository
}

func (c *CanalBinlogConsumer) Consume(ctx context.Context, msg kafka.Message) error {
    // 步骤 1：反序列化 binlog 消息
    var event BinlogEvent
    if err := json.Unmarshal(msg.Value, &event); err != nil {
        return err
    }
    // 步骤 2：只处理 users 表（共用 topic 时的路由）
    if event.Table != "users" {
        return nil
    }
    // 步骤 3：按类型更新 / 删除缓存
    switch event.Type {
    case "INSERT", "UPDATE":
        for _, row := range event.Data {
            uid, _ := strconv.ParseInt(row["id"], 10, 64)  // 列值是字符串！
            user := domain.User{ID: uid, Nickname: row["nickname"], Email: row["email"]}
            _ = c.cache.Set(ctx, user)                      // 直接用 binlog 数据回写
        }
    case "DELETE":
        for _, row := range event.Data {                    // 被删行在 data！
            uid, _ := strconv.ParseInt(row["id"], 10, 64)
            _ = c.cache.Del(ctx, uid)
        }
    }
    return nil
}

案例二：借助 Canal 完成数据校验

第 11 章做不停机迁移时，你要持续核对「源库和目标库的数据一不一致」。当时靠定时全量比对——费时费力。现在有了 Canal：源库每变一行，binlog 消息就告诉你「这个主键的行变了」，只需按主键去两边各读一次、比对、修复——这就是增量校验：

// internal/consumer/binlog_verify.go
// 不管增删改，只拿主键去校验两边
func (c *CanalVerifyConsumer) Consume(ctx context.Context, msg kafka.Message) error {
    var event BinlogEvent
    if err := json.Unmarshal(msg.Value, &event); err != nil {
        return err
    }
    for _, row := range event.Data {
        id, _ := strconv.ParseInt(row["id"], 10, 64)
        if c.useSourceAsTruth {          // 策略一：源库为准
            src, err := c.srcRepo.FindByID(ctx, id)
            if err != nil { return err }
            _ = c.dstRepo.Upsert(ctx, src)    // 把目标库修成和源库一样
        } else {                          // 策略二：目标库为准
            dst, err := c.dstRepo.FindByID(ctx, id)
            if err != nil { return err }
            _ = c.srcRepo.Upsert(ctx, dst)    // 把源库修成和目标库一样
        }
    }
    return nil
}

「以谁为准」切换的陷阱。迁移是有阶段的：切换前「源库为准」（源是真相，修复目标库），切读之后「目标库为准」（目标成了真相，反过来修源库）。切换的那一瞬间埋着一颗雷：

1. 双写阶段，SRC 为准：业务更新 SRC.a = 2，产生 binlog
2. binlog 到达 Canal 时，恰好切换为「DST 为准」
3. 消费者收到 binlog(a=2)，发现是源库的变更，按「目标表为准」策略忽略
4. 实际上 SRC 已经是 a=2，DST 还是 a=1——这条不一致永远没人修了

解决方案：切换期间被修改过的数据需要手动再校验一遍（比如记录切换窗口内变更过的主键清单，补一轮核对）。这个坑很隐蔽，是「讲出来面试官点头」级别的细节。

小收口：原理、消息、两个案例都齐了。三个核心概念收拢成一句话：日志链路让「发生了什么」可检索，伪装从库让「变更」自动流出数据库，增量同步把变更流变成缓存、索引、校验的实时燃料。下面进入动手环节——把两条链路真正跑起来。

四、跟着写：分步实现

上一章我们把搜索服务跑了起来：发帖走 gRPC 双写 ES，SavePost 用 post_id 当文档 ID 实现 Upsert。本章补两块基础设施：日志侧，服务把 JSON 日志写进文件，经 Filebeat → Logstash → ES → Kibana 集中检索；数据侧，Canal 订阅 binlog 发到 Kafka，消费者把变更实时同步进 ES——第 17 章的「业务双写」从此多了一条「数据库自动流」的并行链路（全量同步保留兜底）。七步总览：

flowchart TD
    S1["第 1 步 日志规范
zap 落文件 + service 字段"] --> S2["第 2 步 部署 Filebeat"]
    S2 --> S3["第 3 步 Logstash 管道
+ 索引模板 / ILM"]
    S3 --> S4["第 4 步 验证日志链路"]
    S4 --> S5["第 5 步 部署 Canal
binlog → canal-events"]
    S5 --> S6["第 6 步 canal-consumer
同步 ES（核心）"]
    S6 --> S7["第 7 步 端到端验证
改库 → 秒级同步 + 死信"]

这张图怎么读：从上到下严格串行的七步，前四步立日志链路——第 1 步规范 zap 日志落文件（是第 2 步 Filebeat 能采到日志的前提），第 2 步部署 Filebeat，第 3 步写 Logstash 管道并配索引模板与 ILM，第 4 步在 Kibana 验证日志可搜；后三步搭数据同步链路——第 5 步部署 Canal 让 binlog 流进 canal-events topic，第 6 步写 canal-consumer 幂等同步 ES，第 7 步端到端验证改库秒级同步与死信兜底。哪一步验证不过，就按这条链回溯上一步排查。

第 1 步：日志规范落地（zap + viper + service 字段）

日志链路的地基是日志本身规范：JSON 单行、时间用 ISO8601、每条带 service 字段（将来 10 个服务的日志混在 Kibana 里，全靠它区分）。第 08 章的 zap 日志是打到控制台的，本步升级成「写文件 + 切片管理」。

先加配置（es、kafka 段前几章已配好，这里只新增两段）：

# configs/config.yaml —— 追加 log / canal 段
log:
  dir: ./logs              # 日志目录；生产为 /var/log/content-community，须与 Filebeat 挂载路径一致
  level: info
canal:
  topic: canal-events
  deadLetterTopic: canal-events-dlq   # 死信 topic：同步失败的消息转这里人工处理
  group: canal-consumer

然后写 logger 包。两个关键点：lumberjack 负责切片（日志文件超 100MB 自动切一个新文件）；zap.Fields 给每条日志自动注入 service 字段：

// pkg/logger/logger.go
package logger

import (
	"fmt"
	"path/filepath"

	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"gopkg.in/natefinch/lumberjack.v2"
)

// Config 对应 configs/config.yaml 的 log 段
type Config struct {
	Dir         string `mapstructure:"dir"`         // 日志目录
	Level       string `mapstructure:"level"`       // debug / info / warn / error
	ServiceName string `mapstructure:"serviceName"` // 决定日志文件名与 service 字段
}

// New 写文件的 zap 日志（JSON 单行 + lumberjack 切片）
func New(cfg Config) (*zap.Logger, error) {
	var level zapcore.Level
	if err := level.UnmarshalText([]byte(cfg.Level)); err != nil {
		return nil, fmt.Errorf("解析日志级别: %w", err)
	}
	enc := zap.NewProductionEncoderConfig()
	enc.EncodeTime = zapcore.ISO8601TimeEncoder // 第 3 步 Logstash 的 date 插件按这个格式取时间
	core := zapcore.NewCore(zapcore.NewJSONEncoder(enc),
		zapcore.AddSync(&lumberjack.Logger{
			Filename:   filepath.Join(cfg.Dir, cfg.ServiceName+".log"),
			MaxSize:    100, // 单文件最大 MB，超过自动切片
			MaxBackups: 5,   // 保留旧文件数
			MaxAge:     30,  // 保留天数
			Compress:   true,
		}), level)
	return zap.New(core, zap.AddCaller(),
		zap.Fields(zap.String("service", cfg.ServiceName))), nil
}

验证：go vet ./... 通过；启动任一服务后 cat logs/payment-service.log，每行都是 JSON 且带 "service":"payment-service" 字段。

第 2 步：部署 Filebeat（filebeat.yml + compose）

链路是「应用写宿主机 ./logs → 挂载进容器 /var/log/content-community → Filebeat 采集」。Filebeat 配置在 3.4 节基础上定了型（filestream 输入 + 发 Logstash）：

# filebeat.yml
filebeat.inputs:
  - type: filestream
    id: content-community          # 固定 id，registry 按它识别数据源
    enabled: true
    paths:
      - /var/log/content-community/*.log
    fields:
      env: dev                     # 环境标签，Kibana 里按环境过滤
    fields_under_root: true

output.logstash:
  hosts: ["logstash:5044"]         # 发给 Logstash，不直写 ES（原因见 3.4 节）

再把 logstash、kibana、filebeat 三个服务追加进 docker compose（ES 用第 17 章部署的那个）：

# docker-compose.yml —— 追加三个服务
services:
  logstash:
    image: docker.elastic.co/logstash/logstash:8.0.0
    volumes:
      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf
    ports:
      - "5044:5044"                # Filebeat 发送数据的端口
    depends_on:
      - elasticsearch

  kibana:
    image: docker.elastic.co/kibana/kibana:8.0.0
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200  # 关键：告诉 Kibana ES 在哪
    ports:
      - "5601:5601"
    depends_on:
      - elasticsearch

  filebeat:
    image: docker.elastic.co/beats/filebeat:8.0.0
    user: root                     # 读宿主机日志文件需要权限
    command: filebeat -e --strict.perms=false
    volumes:
      - ./filebeat.yml:/usr/share/filebeat/filebeat.yml:ro
      - ./logs:/var/log/content-community:ro       # 对应第 1 步的 log.dir
      - filebeat-data:/usr/share/filebeat/data     # registry 持久化（3.4 节的坑）
    depends_on:
      - logstash

volumes:
  filebeat-data:

验证：docker compose up -d logstash kibana filebeat 后 docker logs filebeat，看到 harvester 开始读文件、连上 logstash:5044 且无报错。

第 3 步：Logstash 管道 + ES 日志索引模板与 ILM

3.3 节讲过管道三段式，这里写正式版管道。四个 filter 动作逐个对应 3.3 节的插件：json 解析（我们的日志是 JSON，不用 Grok）、date 时间标准化（用日志产生时间，防 Filebeat 积压漂移）、drop 噪声（debug 级别 + 健康检查探活）、mutate 清场（删原始报文和 Filebeat 元字段，Kibana 里干净）：

# logstash.conf
input {
  beats { port => 5044 }                 # 接收 Filebeat 数据
}

filter {
  # 1) zap 的 JSON 单行日志在 message 里，解析后字段挂在 data.* 下
  json { source => "message" target => "data" }

  # 2) 用日志产生时间替换 @timestamp，避免 Filebeat 积压/重启后时间漂移
  date {
    match => ["[data][ts]", "ISO8601"]   # 对应第 1 步的 ISO8601TimeEncoder
    target => "@timestamp"
  }

  # 3) 丢弃噪声：debug 级别 + 健康检查探活日志
  if [data][level] == "debug" or [data][path] == "/health" { drop { } }

  # 4) 清掉原始报文和 Filebeat 元字段
  mutate { remove_field => ["message", "agent", "ecs", "input", "log"] }
}

output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "content-community-logs-%{+YYYY.MM.dd}"  # 按天分索引（原因见 3.6 节）
    manage_template => false                           # 索引模板下面手动建，不让 Logstash 自建
  }
}

验证：docker compose restart logstash 后 docker logs --tail 30 logstash，看到 Pipeline started 且无配置报错。

然后按 3.6 节的思想落地 ILM 和索引模板（30 天转暖合并段、90 天删除；字段显式定型——keyword 才能过滤聚合，long/integer 才能比大小，这是第 17 章 mapping 课的复习）：

# 1) ILM 策略：30 天转暖（合并段），90 天整体删除
curl -X PUT "localhost:9200/_ilm/policy/content-community-logs" \
  -H 'Content-Type: application/json' -d \
  '{"policy":{"phases":{"hot":{"actions":{"set_priority":{"priority":100}}},"warm":{"min_age":"30d","actions":{"forcemerge":{"max_num_segments":1}}},"delete":{"min_age":"90d","actions":{"delete":{}}}}}}'

# 2) 索引模板：匹配按天索引，绑定 ILM + 字段定型
curl -X PUT "localhost:9200/_index_template/content-community-logs" \
  -H 'Content-Type: application/json' -d '
{
  "index_patterns": ["content-community-logs-*"],
  "template": {
    "settings": { "number_of_shards": 1, "number_of_replicas": 0,
                  "index.lifecycle.name": "content-community-logs" },
    "mappings": { "properties": {
      "@timestamp": { "type": "date" }, "env": { "type": "keyword" },
      "data.level": { "type": "keyword" }, "data.service": { "type": "keyword" },
      "data.trace_id": { "type": "keyword" }, "data.user_id": { "type": "long" },
      "data.method": { "type": "keyword" }, "data.path": { "type": "keyword" },
      "data.status": { "type": "integer" }, "data.duration": { "type": "long" },
      "data.message": { "type": "text" }, "data.error": { "type": "text" }
    } }
  }
}'

验证：curl "localhost:9200/_index_template/content-community-logs?pretty" 能查回刚建的模板。

第 4 步：验证日志链路（写一条日志 → Kibana 搜到）

整条链路逐段验证——分段排障是这条链路最重要的排障思路，哪一段断了查哪段：

# 段 1：应用确实写了文件（第 1 步的产物）
tail -1 logs/payment-service.log

# 段 2-4：起全链路，触发一条业务日志（登录失败会打 error）
docker compose up -d
curl -X POST localhost:8080/api/login -H 'Content-Type: application/json' \
  -d '{"email":"bad@test.com","password":"wrong"}'

# 段 5：日志进了 ES —— 按天索引出现且 docs.count > 0
curl "localhost:9200/_cat/indices/content-community-logs-*?v"

最后一段（段 6）在浏览器里完成：打开 http://localhost:5601 → Stack Management → Index Patterns → 创建 content-community-logs-*、时间字段选 @timestamp → 进 Discover。试三类查询，体会「结构化日志 + ES」的威力：

1. data.level: "error"——只看错误；
2. data.duration > 1000——只看慢请求（1 秒以上）；
3. 搜某个 data.trace_id——把一次请求在多个服务里的日志串成一条链（第 08 章埋的 trace_id 在这里兑现）。

第 5 步：Canal 部署并验证（binlog → canal-events）

先把 MySQL 侧准备好。binlog 开 ROW 模式的 compose 参数在 3.8 节给过了（--binlog-format=ROW + --binlog-row-image=FULL）；还需要一个 canal 专用账号——最小权限：只需要读权限 + 复制权限，一丁点写权限都不给：

-- init.sql：只给订阅 binlog 必需的权限
CREATE USER IF NOT EXISTS 'canal'@'%' IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

Canal 的配置分两层（3.7 节组件表里的「配置文件」），只改关键项、其余保持镜像默认：

# conf/canal/canal.properties —— 转发模式：投递到 Kafka（不用 Canal Client 拉模式）
canal.serverMode = kafka
canal.mq.servers  = kafka:9092

# conf/canal/example/instance.properties —— 监听哪个库、发到哪个 topic
canal.instance.master.address = mysql:3306
canal.instance.dbUsername      = canal
canal.instance.dbPassword      = canal
canal.instance.filter.regex    = content_community\\..*   # 只监听这个库的所有表
canal.mq.topic                 = canal-events
canal.mq.partition             = 0
# 需要同主键消息有序时，改按主键 hash 分区（见 3.9 节）并调大 topic 分区数：
# canal.mq.partitionHash = .*\\..*:$pk$

compose 里 canal 服务挂载这两个配置文件（挂到 /home/admin/canal-server/conf/ 下同名路径），依赖 mysql 和 kafka。验证部署：

docker compose up -d canal
docker logs canal        # 预期看到 start the canal server

# 终端 1：订阅 canal-events（kafka-console-consumer 安装见第 07 章）
kafka-console-consumer --bootstrap-server localhost:9092 --topic canal-events

# 终端 2：改一行 post
docker exec -it mysql mysql -uroot -proot content_community \
  -e "UPDATE post SET title='Canal 测试' WHERE id=1;"

终端 1 应立刻吐出（对照 3.8 节的消息格式看：列值全是字符串、UPDATE 带 old）：

{"data":[{"id":"1","title":"Canal 测试","status":"2"}],
 "database":"content_community","table":"post","type":"UPDATE","old":[{"title":"旧标题"}]}

第 6 步：canal-consumer 消费者（本章核心代码）

现在 binlog 消息已经躺在 canal-events topic 里了，还差一个消费者把它同步进 ES。分五小块写：消息结构体 → 消费骨架 → 单条处理与路由 → ES 同步（幂等核心）→ 死信转发。

6.1 消息结构体——字段对照 3.8 节的消息格式，行数据用 map[string]string 接（列值全是字符串的坑）：

// search-service/internal/canalconsumer/handler.go
package canalconsumer

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/IBM/sarama"
	"go.uber.org/zap"

	"github.com/yourname/content-community/search-service/internal/repository"
	pes "github.com/yourname/content-community/search-service/pkg/es"
)

const postIndex = "post" // 与第 17 章 repository 的索引名一致

// Event Canal 发到 Kafka 的 binlog 消息（格式见 3.8 节）
type Event struct {
	Data     []map[string]string `json:"data"` // 变更后的行；DELETE 时放的是被删行（3.8 节的坑）
	Old      []map[string]string `json:"old"`  // UPDATE 时的旧值
	Database string              `json:"database"`
	Table    string              `json:"table"`
	Type     string              `json:"type"` // INSERT / UPDATE / DELETE
	IsDDL    bool                `json:"isDdl"`
	TS       int64               `json:"ts"` // 变更发生的 Unix 秒
}

6.2 消费骨架——消费者组写法与第 07 章一致（sarama.ConsumerGroupHandler 接口），重点是 ConsumeClaim 里的死信策略：同步失败的消息转死信 topic 留底，然后照样提交位点——失败不阻塞主流程：

// search-service/internal/canalconsumer/handler.go（续）
// SyncHandler 按 database+table 路由，post 表变更同步 ES
type SyncHandler struct {
	ES              *pes.Client
	Logger          *zap.Logger
	DLQ             sarama.SyncProducer // 死信生产者：重试无果的消息转人工处理
	DeadLetterTopic string
}

func (h *SyncHandler) Setup(sarama.ConsumerGroupSession) error   { return nil }
func (h *SyncHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }

func (h *SyncHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		if err := h.handle(sess.Context(), msg.Value); err != nil {
			// 失败不阻塞主流程：原始消息转死信 topic 留底
			h.Logger.Error("同步失败，转死信",
				zap.Int32("partition", msg.Partition),
				zap.Int64("offset", msg.Offset), zap.Error(err))
			_ = h.toDeadLetter(msg)
		}
		sess.MarkMessage(msg, "") // 无论成败都提交位点：失败消息已在死信留底
	}
	return nil
}

6.3 单条处理与路由——解析、跳过 DDL（建表改表消息不是数据）、监控同步延迟、按表分发。要同步新表就加一个 case：

// search-service/internal/canalconsumer/handler.go（续）
// handle 单条消息：解析 → 路由 → 分发
func (h *SyncHandler) handle(ctx context.Context, value []byte) error {
	var ev Event
	if err := json.Unmarshal(value, &ev); err != nil {
		return fmt.Errorf("解析 Canal 消息: %w", err)
	}
	if ev.IsDDL { // 建表/改表消息不进 ES
		return nil
	}
	// 同步延迟监控：消息时间戳 vs 当前时间，超 1 分钟告警
	if lag := time.Now().Unix() - ev.TS; lag > 60 {
		h.Logger.Warn("Canal 同步延迟超过 1 分钟",
			zap.Int64("lag_seconds", lag), zap.String("table", ev.Table))
	}
	// 按 database + table 路由，要同步新表就加一个 case
	switch ev.Database + "." + ev.Table {
	case "content_community.post":
		return h.syncPost(ctx, ev)
	default:
		return nil // 其它表暂不同步；user/comment 表按同样套路扩展
	}
}

6.4 ES 同步（幂等核心）——文档 _id 一律用 post_id：INSERT/UPDATE 覆盖写、DELETE 按 _id 删。消息重复消费，最终结果不变——这是 3.7 节「消费者必须幂等」的落地：

// search-service/internal/canalconsumer/handler.go（续）
// syncPost post 表变更 → ES。幂等核心：文档 _id 一律用 post_id
func (h *SyncHandler) syncPost(ctx context.Context, ev Event) error {
	for _, row := range ev.Data {
		id, err := strconv.ParseInt(row["id"], 10, 64) // Canal 列值全是字符串！
		if err != nil {
			return fmt.Errorf("解析 post id: %w", err)
		}
		switch ev.Type {
		case "INSERT", "UPDATE":
			body, err := json.Marshal(toPostDoc(row))
			if err != nil {
				return fmt.Errorf("序列化文档: %w", err)
			}
			resp, err := h.ES.Index(postIndex, h.ES.Index.WithContext(ctx),
				h.ES.Index.WithDocumentID(strconv.FormatInt(id, 10)), // 幂等：文档 ID = post id
				h.ES.Index.WithRefresh("true"),                       // 联调立即可搜，生产去掉
				h.ES.Index.WithBody(bytes.NewReader(body)))
			if err != nil {
				return fmt.Errorf("写入 ES: %w", err)
			}
			resp.Body.Close()
			if resp.IsError() {
				return fmt.Errorf("写入 ES 响应异常: %s", resp.String())
			}
		case "DELETE": // 被删行在 data 里，不在 old（见 3.8 节）
			resp, err := h.ES.Delete(postIndex, strconv.FormatInt(id, 10),
				h.ES.Delete.WithContext(ctx), h.ES.Delete.WithRefresh("true"))
			if err != nil {
				return fmt.Errorf("删除 ES 文档: %w", err)
			}
			resp.Body.Close() // 404（文档已不存在）也视为成功——幂等
		}
	}
	h.Logger.Info("post 表同步完成",
		zap.String("type", ev.Type), zap.Int("rows", len(ev.Data)))
	return nil
}

6.5 行数据转文档 + 死信转发——toPostDoc 把字符串列值转回第 17 章的 PostDoc 结构（注意 datetime 格式要补 T）；toDeadLetter 原样转发到死信 topic，key 带上分区和 offset 方便人工定位：

// search-service/internal/canalconsumer/handler.go（续）
// toPostDoc Canal 行数据 → 第 17 章的 ES 文档结构
func toPostDoc(row map[string]string) repository.PostDoc {
	doc := repository.PostDoc{
		PostID: toInt64(row["id"]), AuthorID: toInt64(row["author_id"]),
		Title: row["title"], Content: row["content"], CoverImage: row["cover_image"],
		Status: int8(toInt64(row["status"])), Visible: int8(toInt64(row["visible"])),
		LikeCount:    int(toInt64(row["like_count"])),
		CommentCount: int(toInt64(row["comment_count"])),
		ViewCount:    int(toInt64(row["view_count"])),
		// MySQL datetime "2026-08-25 15:04:05" → ES date 认的 "2026-08-25T15:04:05"
		CreatedAt: strings.Replace(row["created_at"], " ", "T", 1),
	}
	if row["tags"] != "" { // tags 列存逗号分隔串，按实际存储调整
		doc.Tags = strings.Split(row["tags"], ",")
	}
	return doc
}

func toInt64(s string) int64 {
	v, _ := strconv.ParseInt(s, 10, 64)
	return v
}

// toDeadLetter 原样转发到死信 topic，人工排查后可重放
func (h *SyncHandler) toDeadLetter(msg *sarama.ConsumerMessage) error {
	_, _, err := h.DLQ.SendMessage(&sarama.ProducerMessage{
		Topic: h.DeadLetterTopic,
		Key:   sarama.StringEncoder(fmt.Sprintf("%s-%d-%d", msg.Topic, msg.Partition, msg.Offset)),
		Value: sarama.ByteEncoder(msg.Value),
	})
	return err
}

最后是组装入口 main.go——配置 + 日志复用本章第 1 步，ES 客户端直接复用第 17 章的 pkg/es（连接管理与启动重试都在里面），post 索引由第 17 章的搜索服务负责建：

// search-service/cmd/canal-consumer/main.go
package main

import (
	"context"
	"fmt"
	"os/signal"
	"syscall"
	"time"

	"github.com/IBM/sarama"
	"github.com/spf13/viper"
	"go.uber.org/zap"

	"github.com/yourname/content-community/pkg/logger"
	"github.com/yourname/content-community/search-service/internal/canalconsumer"
	pes "github.com/yourname/content-community/search-service/pkg/es"
)

func main() {
	// 1. 配置 + 日志（复用第 1 步的 logger，service 字段标明身份）
	viper.SetConfigFile("configs/config.yaml")
	if err := viper.ReadInConfig(); err != nil {
		panic(fmt.Errorf("读取配置: %w", err))
	}
	var logCfg logger.Config
	_ = viper.UnmarshalKey("log", &logCfg)
	logCfg.ServiceName = "canal-consumer"
	lg, err := logger.New(logCfg)
	if err != nil {
		panic(err)
	}
	setupConsume(lg)
}

// search-service/cmd/canal-consumer/main.go（续）
// setupConsume 组装依赖：ES 客户端、消费组、死信生产者，然后进入消费循环
func setupConsume(lg *zap.Logger) {
	// 2. ES 客户端：复用第 17 章 pkg/es（含带重试的健康检查）
	var esCfg pes.Config
	_ = viper.UnmarshalKey("es", &esCfg)
	esClient, err := pes.NewClient(esCfg, lg)
	if err != nil {
		panic(err)
	}

	// 3. Kafka：一个消费组 + 一个死信生产者
	brokers := viper.GetStringSlice("kafka.addresses")
	sc := sarama.NewConfig()
	sc.Consumer.Return.Errors = true
	cg, err := sarama.NewConsumerGroup(brokers, viper.GetString("canal.group"), sc)
	if err != nil {
		panic(fmt.Errorf("创建消费组: %w", err))
	}
	dlq, err := sarama.NewSyncProducer(brokers, sc)
	if err != nil {
		panic(fmt.Errorf("创建死信生产者: %w", err))
	}
	handler := &canalconsumer.SyncHandler{
		ES: esClient, Logger: lg, DLQ: dlq,
		DeadLetterTopic: viper.GetString("canal.deadLetterTopic"),
	}
	consumeLoop(cg, handler, lg) // 见下方代码块
}

// search-service/cmd/canal-consumer/main.go（续）
// consumeLoop 循环消费，Ctrl+C 优雅退出
func consumeLoop(cg sarama.ConsumerGroup, handler *canalconsumer.SyncHandler, lg *zap.Logger) {
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()
	brokers := viper.GetStringSlice("kafka.addresses")
	lg.Info("canal-consumer 启动", zap.Strings("brokers", brokers))
	for {
		if err := cg.Consume(ctx, []string{viper.GetString("canal.topic")}, handler); err != nil {
			lg.Error("消费异常，5 秒后重试", zap.Error(err))
			select {
			case <-time.After(5 * time.Second):
			case <-ctx.Done():
				return
			}
		}
		if ctx.Err() != nil {
			return
		}
	}
}

验证：go vet ./... 通过；go run ./search-service/cmd/canal-consumer 启动无报错，日志打出 canal-consumer 启动。

第 7 步：端到端验证（改库 → ES 秒级同步）

三个终端配合，把整条链路跑闭环：

# 终端 1：启动消费者
go run ./search-service/cmd/canal-consumer

# 终端 2：改一行 post
docker exec -it mysql mysql -uroot -proot content_community \
  -e "UPDATE post SET title='端到端验证' WHERE id=1;"

# 终端 3：查 ES（文档 ID = post id，1 秒内应更新）
curl -s "localhost:9200/post/_doc/1?pretty" | grep title

预期输出 "title" : "端到端验证"，终端 1 打出 post 表同步完成 {"type": "UPDATE", "rows": 1}。再补两刀验证完整性：

# 验证删除：删行后 ES 文档同步消失
docker exec -it mysql mysql -uroot -proot content_community \
  -e "DELETE FROM post WHERE id=1;"
curl -s "localhost:9200/post/_doc/1?pretty"   # 预期 404

# 验证死信：停掉 ES 再改库，失败消息应出现在 canal-events-dlq
docker compose stop elasticsearch
docker exec -it mysql mysql -uroot -proot content_community \
  -e "UPDATE post SET title='dlq 测试' WHERE id=2;"
kafka-console-consumer --bootstrap-server localhost:9092 --topic canal-events-dlq --from-beginning

「改库 → binlog → canal-events → 消费者 → ES」链路闭环，失败有死信兜底。恢复 ES 后把死信消息重放（再发回原 topic 或写个重放工具），ES 会追平。

本章完成后的项目结构

content-community/
├── configs/config.yaml                    # 修改：新增 log / canal 段
├── docker-compose.yml                     # 修改：追加 logstash、kibana、filebeat、canal 服务
├── filebeat.yml                           # 新增：第 2 步
├── logstash.conf                          # 新增：第 3 步
├── init.sql                               # 修改：追加 canal 账号授权
├── conf/canal/                            # 新增：canal.properties + example/instance.properties（第 5 步）
├── pkg/logger/logger.go                   # 新增：第 1 步（viper 化 zap 初始化）
└── search-service/
    ├── cmd/canal-consumer/main.go         # 新增：第 6 步（消费入口）
    └── internal/canalconsumer/handler.go  # 新增：第 6 步（路由 + ES 同步 + 死信）

验证清单

• 服务日志写入 logs/<service>.log，每行是带 service 字段的 JSON 单行（第 1 步）
• curl "localhost:9200/_cat/indices/content-community-logs-*?v" 出现按天索引；Kibana 能按 data.level、data.duration、data.trace_id 查日志（第 4 步）
• kafka-console-consumer 订阅 canal-events，改 post 表能看到 UPDATE 消息（第 5 步）
• canal-consumer 启动无报错；改库后 1 秒内 ES 文档更新，删行后 _doc/<id> 返回 404（第 7 步）
• 停掉 ES 再改库，失败消息进 canal-events-dlq；同一条消息重放两次，ES 结果不变（幂等验证）

五、工程实践要点

1. 日志规范在源头，别在 Logstash 里补救

解析逻辑越靠近源头（代码里的 zap 结构化日志），链路上每一环越简单。Logstash 的 filter 只做「格式微调」（时间标准化、删噪声、删元字段），不做业务逻辑——否则那份 conf 会变成没人敢动的祖传资产。配套纪律：全程 JSON 单行，别让 fmt.Println 的裸文本混进日志文件。

2. 三个「书签」是一个思想，但都不保证下游成功

Filebeat 的 registry、Canal 的位点、Kafka 消费者的 offset，都是「记住处理到哪了」的断点续传设计。但要清醒：前两个记录的是「我发出去 / 投递出去」，不是「下游处理成功」。所以链路上最后一个消费者永远承担幂等责任——覆盖写天然幂等，追加型操作（计数 +1）必须自己去重。

3. 死信是「不阻塞」的前提，不是摆设

消费失败的消息转死信 + 提交位点，换来的是主流程永不卡死。但死信 topic 必须有人管：监控堆积量、定期人工排查重放。只建不修，等于把故障从「实时阻断」变成「慢性失血」。

4. 日志保留策略必须先于磁盘告警存在

按天分索引 + ILM 是上线当天就要配的事，不是「等日志多了再说」。磁盘水位 85% 分片迁移、95% 索引只读——那是凌晨三点级别的告警。本地学习环境至少定期手动清旧索引。

5. Canal 更新缓存是「缓存策略」，不是业务逻辑

binlog 消费者绕开 Service 层直接操作 Repository / Cache——它维护的是缓存这份「副本」，不执行业务规则。把 Service 依赖拖进消费者，等于让缓存同步链路背上业务变更的风险。

6. canal 账号最小权限

只给 SELECT, REPLICATION SLAVE, REPLICATION CLIENT——Canal 只需要「读 binlog」，一丁点写权限都不该有。同步账号泄漏的最坏后果被限制在「能读」，而不是「能改」。

7. 监控同步延迟

消费者里比较「消息时间戳 vs 当前时间」，延迟超阈值告警（第 6 步代码里的 lag 检查）。同步链路最大的隐患不是报错（有死信），而是静默延迟——用户改了帖子，搜索十分钟搜不到旧标题，还没人知道。

8. 表与 topic 的规模化路线

小规模共用一个 topic（按 table 字段路由，本课程做法）；流量上来再拆：大表独立 topic、甚至独立 Kafka 集群。别等一张大表的变更洪峰把全公司的同步消息都堵住才动手。

六、自测题与动手练习

自测题（合上书能答出来才算懂；答案都在正文对应小节里）：

1. ELK 的四段式架构是哪四段？Logstash 的三段式（Input / Filter / Output）和它是什么关系？（3.1 / 3.3 节）
2. 为什么生产上要 Filebeat 蹲机器、Logstash 集中处理，而不是每台机器跑一个 Logstash？（3.4 节）
3. Logstash 的 date 插件为什么要把日志自带时间写进 @timestamp，不这么干会怎样？（3.3 节）
4. Canal 怎么「骗过」MySQL 拿到 binlog？说出复制协议的四步。（3.7 节）
5. binlog 的 Row 和 Statement 模式各记什么？为什么 Canal 场景必须 Row？（3.8 节）
6. Canal 的 DELETE 消息里，被删数据在 data 还是 old？列值是什么类型？（3.8 节）
7. Canal 更新缓存为什么必须按主键 hash 分区？同主键消息分散到不同分区会发生什么？（3.9 节）
8. 为什么说「Canal 的位点只保证投递，不保证消费」？由此推出的消费者设计原则是什么？（3.7 / 第 6 步）

动手练习（建议真做一遍，难度递增）：

1. 起一套 ELK + Filebeat（对应第 1–4 步）：把自己任一 Go 服务的 JSON 日志配进链路，在 Kibana Discover 里完成三类查询（按 level、按 duration、按 trace_id）。做到什么程度算完成：故意把 Logstash 的 json 配置删掉重启，Kibana 里出现解析失败现象，再改回来恢复——你能解释失败时日志去哪了。
2. 起 Canal 看 binlog（对应第 5 步）：改一行 users 表，用 kafka-console-consumer 看到消息，并手动比对 UPDATE 消息里 data 和 old 的差异。做到什么程度算完成：能指出消息里任意一个字段（database / table / type / ts / isDdl）的含义。
3. 写个缓存同步消费者（对应第 6 步思路 + 3.9 节）：监听 binlog，收到 users 表变更就更新一个本地 Map 缓存，验证「改库 → 缓存自动变」。做到什么程度算完成：把 canal 的分区配置改成单分区再压一轮并发更新，观察顺序是否仍然正确，并解释原因。

七、模拟面试问答

以下 12 道题覆盖本章所有高频考点，按「基础 → 进阶 → 追问」三档排列。每题给出可以直接背的一句话结论、展开思路、面试官想听的关键词、新手最容易答错的地方。

八、本章小结

下一章（第 19 章）我们进入 Feed 流设计与压测——内容社区最复杂的业务之一：百万粉丝博主的「发布」怎样才能不把系统写炸、普通用户的「刷新」怎样才能 50ms 内拉完？你会把前面积累的积木（Kafka 异步、Redis 缓存、聚合接口、推拉模型）全部用上，搭一个推拉结合的 Feed 流，再用 k6 压测把它的性能拐点压出来。本章的 Kafka、顺序性、异步削峰会全部再次登场——准备好了就翻页。