未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

warning 适用边界与维护说明

本文适用于具备 Kafka、Flink 基础的数据开发工程师和架构师，开展假设性电商实时平台的分层设计、磁盘容量估算、积压排查及灾备演练方案讨论，不是已验证的企业案例或可直接部署的完整工程；本次仅作全文结构审阅，未作外部事实核验或运行验证。Filebeat 片段未生成规范要求的全部字段，${timestamp} 仅为环境变量替换；Flink 示例仍需补齐状态初始化、TTL、Checkpoint、配套类型及下游实现。普通消费组查询、Lag 告警和灾备等待循环均已使用第 6 列 LAG，灾备循环也已处理查询失败、无有效记录和不可用位点，不应再引用旧说明中的相关错误。灾备 Step 1 仍在备端查询未经确认的硬编码复制组，不能据此确认同步完成；切换前须核对实际源端复制组、目标 Topic、数据复制结果和消费位点映射，Step 6 的日期输出也不等于实际 RTO。全链路图中的 GMV 数据来源尚未与订单支付链路衔接，宜作为待完善的设计草图。容量模型支持存储估算，不能独立推导 Broker 数量；分区、吞吐、告警阈值及容量余量须结合具体负载确定，ZooKeeper 建议仅适用于相应部署模式。RPO/RTO 为待实测或演练的目标，已取消跳转的两篇延伸材料不阻断现有练习。已取消无法确定目标的可点击引用「03-Flink与Kafka集成」，保留原提及；Missing link target: 03-Flink与Kafka集成已取消无法确定目标的可点击引用「01-ClickHouse概述」，保留原提及；Missing link target: 01-ClickHouse概述

企业级项目实战 - Kafka综合应用案例

abstract 本节概览

学习目标：通过企业级项目掌握 Kafka 综合应用，建立从架构设计到生产运维的完整视角
前置知识：Kafka 最佳实践
预计用时：60 分钟
后续学习：Flink 流处理与 Kafka 深度联动

项目背景：电商实时数据平台

业务需求分析

本项目采用假设性电商平台场景：日活跃用户 5000 万，峰值每秒产生 120 万条事件，涉及用户行为、订单支付、库存变更等十几个业务系统；这些数值是设计输入，不代表已验证的企业实际数据。核心诉求：实时感知业务状态，分钟级驱动运营决策。

%%{init: {"theme":"base","themeVariables":{"fontFamily":"PingFang SC, -apple-system, Source Han Serif SC, serif","fontSize":"16px","background":"#FDFBF5","primaryColor":"#FDFBF5","primaryTextColor":"#2D2A26","primaryBorderColor":"#6E6A60","secondaryColor":"#E8EFE8","secondaryBorderColor":"#8FA890","secondaryTextColor":"#2D2A26","tertiaryColor":"#FDFBF5","tertiaryBorderColor":"#9A9488","tertiaryTextColor":"#6E6A60","lineColor":"#6E6A60","textColor":"#2D2A26","mainBkg":"#FDFBF5","nodeBorder":"#2D5A3D","clusterBkg":"#F3EEDF","clusterBorder":"#9A9488","edgeLabelBackground":"#FDFBF5"},"flowchart":{"htmlLabels":true,"curve":"basis","padding":20,"nodeSpacing":60,"rankSpacing":70,"useMaxWidth":false},"themeCSS":".node rect, .node polygon, .node circle { stroke-width: 1.5px; } .edgePath .path { stroke-width: 1.5px; } .cluster-label .nodeLabel, .cluster-label foreignObject { color: #2D2A26 !important; } .edgeLabel, .edgeLabel foreignObject { color: #2D2A26 !important; background-color: #FDFBF5; } .edgeLabel rect { fill: #FDFBF5; }"}}%%
flowchart TD
    classDef default fill:#FDFBF5,stroke:#6E6A60,color:#2D2A26,stroke-width:1.5px
    classDef startend fill:#2D2A26,stroke:#2D2A26,color:#FDFBF5,font-weight:bold
    classDef decision fill:#F6EAD8,stroke:#C08A1A,color:#2D2A26,font-weight:bold
    classDef hub fill:#3B5A80,stroke:#3B5A80,color:#FDFBF5,font-weight:bold
    classDef io fill:#E5EAF0,stroke:#3B5A80,color:#2D2A26
    classDef accent fill:#A6432C,stroke:#A6432C,color:#FDFBF5,font-weight:bold
    classDef storage fill:#F3EEDF,stroke:#9A9488,color:#2D2A26
    SRC["数据源层\n用户埋点 / 订单系统 / 支付系统\nMySQL Binlog / IoT 设备 / 第三方 API"]
    KAFKA["Kafka 消息层\nODS Topic（原始） / DWD Topic（明细）\nADS Topic（聚合指标）"]
    PROC["Flink 实时计算层\n数据清洗 / 维度关联\n窗口聚合 / 实时指标"]
    STORE["存储应用层\nHBase（用户画像）/ Redis（实时大屏）\nClickHouse（历史分析）/ MySQL（业务回写）"]
    SRC --> KAFKA --> PROC --> STORE
    class SRC,KAFKA,PROC,STORE default

实时数仓全链路架构

数据分层与 Topic 命名规范

实时数仓复用离线数仓的分层理念，每一层对应独立的 Kafka Topic，层间由 Flink 作业驱动流转。

层次	职责	Topic 命名规范	分区数	保留时长
ODS（原始层）	存储原始事件，不做任何加工	`ods.{业务域}.{表名}`	与上游并发对齐	7 天
DWD（明细层）	清洗、脱敏、维度关联后的明细事件	`dwd.{业务域}.{主题}`	与 Flink 并发对齐	3 天
DWS（汇总层）	按业务维度的轻度聚合结果	`dws.{业务域}.{粒度}`	较少，通常 4-8	1 天
ADS（应用层）	面向大屏、接口的指标推送	`ads.{应用}.{指标}`	1-4（低并发消费）	1 小时

# ODS 层 Topic 创建示例（原始用户事件）
kafka-topics.sh --bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092 \
    --create --topic ods.user.behavior_event \
    --partitions 24 --replication-factor 3 \
    --config retention.ms=604800000 \
    --config compression.type=lz4 \
    --config min.insync.replicas=2

# DWD 层 Topic（清洗后的用户行为明细）
kafka-topics.sh --bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092 \
    --create --topic dwd.user.behavior_detail \
    --partitions 24 --replication-factor 3 \
    --config retention.ms=259200000 \
    --config compression.type=lz4 \
    --config min.insync.replicas=2

# ADS 层 Topic（实时 GMV 指标）
kafka-topics.sh --bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092 \
    --create --topic ads.dashboard.realtime_gmv \
    --partitions 4 --replication-factor 3 \
    --config retention.ms=3600000 \
    --config compression.type=lz4

数据源到 ODS 的接入方式

%%{init: {"theme":"base","themeVariables":{"fontFamily":"PingFang SC, -apple-system, Source Han Serif SC, serif","fontSize":"16px","background":"#FDFBF5","primaryColor":"#FDFBF5","primaryTextColor":"#2D2A26","primaryBorderColor":"#6E6A60","secondaryColor":"#E8EFE8","secondaryBorderColor":"#8FA890","secondaryTextColor":"#2D2A26","tertiaryColor":"#FDFBF5","tertiaryBorderColor":"#9A9488","tertiaryTextColor":"#6E6A60","lineColor":"#6E6A60","textColor":"#2D2A26","mainBkg":"#FDFBF5","nodeBorder":"#2D5A3D","clusterBkg":"#F3EEDF","clusterBorder":"#9A9488","edgeLabelBackground":"#FDFBF5"},"flowchart":{"htmlLabels":true,"curve":"basis","padding":20,"nodeSpacing":60,"rankSpacing":70,"useMaxWidth":false},"themeCSS":".node rect, .node polygon, .node circle { stroke-width: 1.5px; } .edgePath .path { stroke-width: 1.5px; } .cluster-label .nodeLabel, .cluster-label foreignObject { color: #2D2A26 !important; } .edgeLabel, .edgeLabel foreignObject { color: #2D2A26 !important; background-color: #FDFBF5; } .edgeLabel rect { fill: #FDFBF5; }"}}%%
flowchart LR
    classDef default fill:#FDFBF5,stroke:#6E6A60,color:#2D2A26,stroke-width:1.5px
    classDef startend fill:#2D2A26,stroke:#2D2A26,color:#FDFBF5,font-weight:bold
    classDef decision fill:#F6EAD8,stroke:#C08A1A,color:#2D2A26,font-weight:bold
    classDef hub fill:#3B5A80,stroke:#3B5A80,color:#FDFBF5,font-weight:bold
    classDef io fill:#E5EAF0,stroke:#3B5A80,color:#2D2A26
    classDef accent fill:#A6432C,stroke:#A6432C,color:#FDFBF5,font-weight:bold
    classDef storage fill:#F3EEDF,stroke:#9A9488,color:#2D2A26
    subgraph 日志类
        FB[Filebeat\n收集 Nginx/App 日志]
        FL[Flume\n多业务线汇聚]
    end
    subgraph 数据库变更
        CANAL[Canal\n监听 MySQL Binlog]
        DT[Debezium\n通用 CDC 方案]
    end
    subgraph 端上埋点
        SDK[JS/App SDK\n行为埋点]
        GW[数据网关\n接入层限流]
    end
    FB --> |ods.infra.nginx_access| K[(Kafka ODS)]
    FL --> |ods.infra.app_log| K
    CANAL --> |ods.order.order_info| K
    DT --> |ods.pay.pay_record| K
    SDK --> GW --> |ods.user.behavior_event| K
    class K storage
    class FB,FL,CANAL,DT,SDK,GW default

日志收集系统：多业务线隔离

Filebeat 到 Kafka 的接入配置片段

日志收集场景的核心挑战：多业务线 Topic 隔离（避免互相干扰）+ 日志格式标准化（方便下游统一处理）。

# filebeat.yml - 接入配置示例，非完整生产配置；使用前需确认 Filebeat 版本及 log input 兼容性
filebeat.inputs:
  # 业务线 A：电商订单服务
  - type: log
    paths:
      - /var/log/order-service/*.log
    fields:
      biz_line: order
      env: production
    fields_under_root: true
    multiline:
      pattern: '^\d{4}-\d{2}-\d{2}'
      negate: true
      match: after

  # 业务线 B：用户行为服务
  - type: log
    paths:
      - /var/log/user-service/*.log
    fields:
      biz_line: user
      env: production
    fields_under_root: true

processors:
  # 标准化日志格式：统一添加采集元数据
  - add_fields:
      target: ''
      fields:
        collector_host: "${HOSTNAME}"
        # ${timestamp} 是环境变量替换，不会逐条生成采集时间；动态采集时间需另行处理
        collect_ts: "${timestamp}"
  # 丢弃空行
  - drop_event:
      when:
        equals:
          message: ""

output.kafka:
  hosts: ["kafka-01:9092", "kafka-02:9092", "kafka-03:9092"]
  # 按业务线路由到不同 Topic
  topic: 'ods.%{[biz_line]}.application_log'
  partition.round_robin:
    reachable_only: true
  required_acks: 1
  compression: lz4
  max_message_bytes: 1048576
  # 批量发送提升吞吐
  bulk_max_size: 2048
  flush_timeout: 1s

日志格式标准化规范

所有接入 ODS 的日志事件必须携带以下标准字段，下游 Flink 作业依赖这些字段进行路由和时间对齐：

{
  "event_id": "uuid-v4",
  "biz_line": "order",
  "env": "production",
  "host": "app-server-01",
  "log_level": "INFO",
  "timestamp": 1700000000000,
  "message": "原始日志内容",
  "collector_host": "filebeat-node-03",
  "collect_ts": 1700000000123
}

warning 时间字段陷阱

timestamp 应使用业务时间（事件发生时间），而非采集时间。Flink 的事件时间窗口依赖此字段，若用采集时间会导致窗口计算结果偏差。

用户行为分析：埋点事件实时链路

端到端架构

%%{init: {"theme":"base","themeVariables":{"fontFamily":"PingFang SC, -apple-system, Source Han Serif SC, serif","fontSize":"16px","background":"#FDFBF5","primaryColor":"#FDFBF5","primaryTextColor":"#2D2A26","primaryBorderColor":"#6E6A60","secondaryColor":"#E8EFE8","secondaryBorderColor":"#8FA890","secondaryTextColor":"#2D2A26","tertiaryColor":"#FDFBF5","tertiaryBorderColor":"#9A9488","tertiaryTextColor":"#6E6A60","lineColor":"#6E6A60","textColor":"#2D2A26","mainBkg":"#FDFBF5","nodeBorder":"#2D5A3D","clusterBkg":"#F3EEDF","clusterBorder":"#9A9488","edgeLabelBackground":"#FDFBF5"},"flowchart":{"htmlLabels":true,"curve":"basis","padding":20,"nodeSpacing":60,"rankSpacing":70,"useMaxWidth":false},"themeCSS":".node rect, .node polygon, .node circle { stroke-width: 1.5px; } .edgePath .path { stroke-width: 1.5px; } .cluster-label .nodeLabel, .cluster-label foreignObject { color: #2D2A26 !important; } .edgeLabel, .edgeLabel foreignObject { color: #2D2A26 !important; background-color: #FDFBF5; } .edgeLabel rect { fill: #FDFBF5; }"}}%%
flowchart TD
    classDef default fill:#FDFBF5,stroke:#6E6A60,color:#2D2A26,stroke-width:1.5px
    classDef startend fill:#2D2A26,stroke:#2D2A26,color:#FDFBF5,font-weight:bold
    classDef decision fill:#F6EAD8,stroke:#C08A1A,color:#2D2A26,font-weight:bold
    classDef hub fill:#3B5A80,stroke:#3B5A80,color:#FDFBF5,font-weight:bold
    classDef io fill:#E5EAF0,stroke:#3B5A80,color:#2D2A26
    classDef accent fill:#A6432C,stroke:#A6432C,color:#FDFBF5,font-weight:bold
    classDef storage fill:#F3EEDF,stroke:#9A9488,color:#2D2A26
    CLIENT["客户端 SDK\n点击 / 曝光 / 搜索 / 加购"]
    GW["数据接入网关\n限流 / 鉴权 / 格式校验"]
    ODS["ods.user.behavior_event\n原始埋点事件"]
    FLINK_DWD["Flink DWD 作业\n会话拼接 / 设备识别 / 脱敏"]
    DWD["dwd.user.behavior_detail\n标准化明细事件"]
    FLINK_ADS["Flink ADS 作业\n实时 UV/PV / 漏斗计算"]
    HBASE["HBase\n用户实时画像"]
    REDIS["Redis\n实时大屏指标"]
    CLIENT --> GW --> ODS --> FLINK_DWD --> DWD
    DWD --> FLINK_ADS --> REDIS
    DWD --> HBASE
    class CLIENT,GW,ODS,FLINK_DWD,DWD,FLINK_ADS,HBASE,REDIS default

Flink 实时计算核心逻辑

// 用户行为实时分析 Flink 作业（核心片段）
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(24);

// 从 Kafka DWD 层读取
KafkaSource<UserBehaviorEvent> source = KafkaSource.<UserBehaviorEvent>builder()
    .setBootstrapServers("kafka-01:9092,kafka-02:9092,kafka-03:9092")
    .setTopics("dwd.user.behavior_detail")
    .setGroupId("flink-ads-user-behavior")
    .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
    .setValueOnlyDeserializer(new UserBehaviorEventDeserializer())
    .build();

DataStream<UserBehaviorEvent> events = env
    .fromSource(source, WatermarkStrategy
        .<UserBehaviorEvent>forBoundedOutOfOrderness(Duration.ofSeconds(10))
        .withTimestampAssigner((e, ts) -> e.getEventTimestamp()), "behavior-source");

// 1 分钟滚动窗口实时 PV/UV
DataStream<PageMetric> pageMetrics = events
    .filter(e -> "PAGE_VIEW".equals(e.getEventType()))
    .keyBy(UserBehaviorEvent::getPageId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new PageMetricAggregator(), new PageMetricWindowFunction());

// 写入 Redis 供大屏消费
pageMetrics.addSink(new RedisSink<>(jedisPoolConfig, new PageMetricMapper()));

// 同时写回 ADS Kafka Topic（供其他系统订阅）
pageMetrics.sinkTo(KafkaSink.<PageMetric>builder()
    .setBootstrapServers("kafka-01:9092,kafka-02:9092,kafka-03:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("ads.dashboard.page_metrics")
        .setValueSerializationSchema(new PageMetricSerializer())
        .build())
    .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
    .build());

订单状态机：Canal Binlog 实时同步

Canal + Kafka + Flink CDC 架构

订单系统的状态变更通过 MySQL Binlog 捕获，经 Canal 转成结构化事件写入 Kafka，再由 Flink 驱动状态机处理。

%%{init: {"theme":"base","themeVariables":{"fontFamily":"PingFang SC, -apple-system, Source Han Serif SC, serif","fontSize":"16px","background":"#FDFBF5","primaryColor":"#FDFBF5","primaryTextColor":"#2D2A26","primaryBorderColor":"#6E6A60","secondaryColor":"#E8EFE8","secondaryBorderColor":"#8FA890","secondaryTextColor":"#2D2A26","tertiaryColor":"#FDFBF5","tertiaryBorderColor":"#9A9488","tertiaryTextColor":"#6E6A60","lineColor":"#6E6A60","textColor":"#2D2A26","mainBkg":"#FDFBF5","nodeBorder":"#2D5A3D","clusterBkg":"#F3EEDF","clusterBorder":"#9A9488","edgeLabelBackground":"#FDFBF5"},"flowchart":{"htmlLabels":true,"curve":"basis","padding":20,"nodeSpacing":60,"rankSpacing":70,"useMaxWidth":false},"themeCSS":".node rect, .node polygon, .node circle { stroke-width: 1.5px; } .edgePath .path { stroke-width: 1.5px; } .cluster-label .nodeLabel, .cluster-label foreignObject { color: #2D2A26 !important; } .edgeLabel, .edgeLabel foreignObject { color: #2D2A26 !important; background-color: #FDFBF5; } .edgeLabel rect { fill: #FDFBF5; }"}}%%
sequenceDiagram
    participant MySQL as MySQL 订单库
    participant Canal as Canal Server
    participant Kafka as Kafka（ods.order.order_info）
    participant Flink as Flink CDC 作业
    participant State as Flink 状态存储（RocksDB）
    participant Sink as 下游系统

    MySQL->>Canal: Binlog 变更事件
    Canal->>Kafka: 序列化为 JSON/Avro 格式
    Kafka->>Flink: 消费事件（at-least-once）
    Flink->>State: 读取订单当前状态
    Flink->>State: 幂等性校验（event_id 去重）
    State->>Flink: 允许/拒绝状态转移
    Flink->>Sink: 写出状态变更结果

Canal 配置要点

# canal.properties - 核心配置项
canal.instance.mysql.slaveId=1234
canal.instance.master.address=mysql-primary:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal_password

# 输出到 Kafka
canal.mq.servers=kafka-01:9092,kafka-02:9092,kafka-03:9092
canal.mq.topic=ods.order.order_info
# 按 order_id 路由，保证同一订单事件有序
canal.mq.partitionsNum=12
canal.mq.partitionHash=.*\\.order_info:order_id

Flink 状态机幂等性保障

// 订单状态机处理（关键：幂等性校验）
public class OrderStateMachineFunction extends KeyedProcessFunction<String, OrderBinlogEvent, OrderStateChange> {

    // 存储订单当前状态
    private ValueState<OrderState> currentState;
    // 已处理事件 ID；本片段未实现 TTL，24 小时去重窗口需在状态描述符或清理逻辑中另行配置
    private MapState<String, Long> processedEventIds;

    @Override
    public void processElement(OrderBinlogEvent event, Context ctx, Collector<OrderStateChange> out) throws Exception {
        // 1. 幂等性校验
        if (processedEventIds.contains(event.getEventId())) {
            return; // 重复事件，直接跳过
        }

        OrderState state = currentState.value();
        String fromStatus = state != null ? state.getStatus() : "INIT";
        String toStatus = event.getNewStatus();

        // 2. 状态机合法性校验
        if (!isValidTransition(fromStatus, toStatus)) {
            // 记录异常，不抛出（避免消费阻塞）
            log.warn("非法状态转移: orderId={}, from={}, to={}", event.getOrderId(), fromStatus, toStatus);
            return;
        }

        // 3. 更新状态
        currentState.update(new OrderState(toStatus, event.getTimestamp()));
        processedEventIds.put(event.getEventId(), event.getTimestamp());

        // 4. 输出状态变更事件
        out.collect(new OrderStateChange(event.getOrderId(), fromStatus, toStatus, event.getTimestamp()));
    }

    private boolean isValidTransition(String from, String to) {
        // 合法的订单状态流转矩阵
        Map<String, Set<String>> transitions = Map.of(
            "INIT",      Set.of("CREATED"),
            "CREATED",   Set.of("PAID", "CANCELLED"),
            "PAID",      Set.of("SHIPPED", "REFUNDING"),
            "SHIPPED",   Set.of("DELIVERED"),
            "DELIVERED", Set.of("COMPLETED", "REFUNDING"),
            "REFUNDING", Set.of("REFUNDED")
        );
        return transitions.getOrDefault(from, Set.of()).contains(to);
    }
}

大规模集群运维

Topic 分区规划原则

分区数是 Kafka 集群最重要的规划决策之一，确定后调整成本极高。

规划维度	建议	说明
分区数规划起点	结合目标消费并行度与实测吞吐确定	同一消费组内的分区消费并行度受分区数限制，生产者数量不构成分区数的硬性下限
分区数上限	Broker 数量 × 100（经验值）	过多分区增加 Controller 和 ZooKeeper 负担
消息量参考	`单分区 10-50 MB/s`	超过此吞吐需增加分区
分区键选择	高基数、分布均匀的业务 ID	避免数据倾斜，`user_id` 好于 `category_id`
扩容时机	消费 Lag 持续增长且无法通过参数调优解决	分区扩容可能改变按 key 的分区路由；需设计迁移与消费衔接，不能默认保证同一业务键跨扩容的顺序

# 查询集群所有 Topic 的分区分布情况
kafka-topics.sh --bootstrap-server kafka-01:9092 --describe | \
  awk '/Topic:/{topic=$2} /PartitionCount:/{print topic, $0}' | \
  sort -k3 -n -r | head -20

# 检查分区 Leader 分布是否均匀
kafka-topics.sh --bootstrap-server kafka-01:9092 --describe | \
  awk '/Leader:/{print $6}' | sort | uniq -c | sort -rn

Consumer Group 管理规范

# 查看所有 Consumer Group 的消费 Lag
kafka-consumer-groups.sh --bootstrap-server kafka-01:9092 \
    --describe --all-groups 2>/dev/null | \
    awk 'NR>1 && $6 ~ /^[0-9]+$/ && $6 > 0 {print $1, $2, $3, "LAG="$6}' | \
    sort -t= -k2,2nr | head -20

# 重置指定 Consumer Group 的 Offset（谨慎操作）
# 场景：Flink 作业重启后需要从指定时间点重新消费
kafka-consumer-groups.sh --bootstrap-server kafka-01:9092 \
    --group flink-ads-user-behavior \
    --topic dwd.user.behavior_detail \
    --reset-offsets --to-datetime 2024-01-15T00:00:00.000 \
    --execute

Lag 告警体系

生产环境必须对 Consumer Lag 建立分级告警，防止消息积压影响业务实时性。

#!/bin/bash
# lag_monitor.sh - Consumer Lag 监控脚本，配合 Prometheus AlertManager 使用

BOOTSTRAP_SERVERS="kafka-01:9092,kafka-02:9092,kafka-03:9092"
LAG_WARN_THRESHOLD=10000      # 告警阈值：积压超过 1 万条
LAG_CRITICAL_THRESHOLD=100000 # 严重阈值：积压超过 10 万条

kafka-consumer-groups.sh --bootstrap-server $BOOTSTRAP_SERVERS \
    --describe --all-groups 2>/dev/null | \
    awk -v warn=$LAG_WARN_THRESHOLD -v critical=$LAG_CRITICAL_THRESHOLD \
    'NR>1 && $6 ~ /^[0-9]+$/ {
        lag = $6 + 0
        if (lag >= critical) {
            printf "CRITICAL: group=%s topic=%s partition=%s lag=%s\n", $1, $2, $3, $6
        } else if (lag >= warn) {
            printf "WARNING: group=%s topic=%s partition=%s lag=%s\n", $1, $2, $3, $6
        }
    }'

灾备方案：MirrorMaker 2 跨机房同步

整体灾备架构

%%{init: {"theme":"base","themeVariables":{"fontFamily":"PingFang SC, -apple-system, Source Han Serif SC, serif","fontSize":"16px","background":"#FDFBF5","primaryColor":"#FDFBF5","primaryTextColor":"#2D2A26","primaryBorderColor":"#6E6A60","secondaryColor":"#E8EFE8","secondaryBorderColor":"#8FA890","secondaryTextColor":"#2D2A26","tertiaryColor":"#FDFBF5","tertiaryBorderColor":"#9A9488","tertiaryTextColor":"#6E6A60","lineColor":"#6E6A60","textColor":"#2D2A26","mainBkg":"#FDFBF5","nodeBorder":"#2D5A3D","clusterBkg":"#F3EEDF","clusterBorder":"#9A9488","edgeLabelBackground":"#FDFBF5"},"flowchart":{"htmlLabels":true,"curve":"basis","padding":20,"nodeSpacing":60,"rankSpacing":70,"useMaxWidth":false},"themeCSS":".node rect, .node polygon, .node circle { stroke-width: 1.5px; } .edgePath .path { stroke-width: 1.5px; } .cluster-label .nodeLabel, .cluster-label foreignObject { color: #2D2A26 !important; } .edgeLabel, .edgeLabel foreignObject { color: #2D2A26 !important; background-color: #FDFBF5; } .edgeLabel rect { fill: #FDFBF5; }"}}%%
flowchart TB
    classDef default fill:#FDFBF5,stroke:#6E6A60,color:#2D2A26,stroke-width:1.5px
    classDef startend fill:#2D2A26,stroke:#2D2A26,color:#FDFBF5,font-weight:bold
    classDef decision fill:#F6EAD8,stroke:#C08A1A,color:#2D2A26,font-weight:bold
    classDef hub fill:#3B5A80,stroke:#3B5A80,color:#FDFBF5,font-weight:bold
    classDef io fill:#E5EAF0,stroke:#3B5A80,color:#2D2A26
    classDef accent fill:#A6432C,stroke:#A6432C,color:#FDFBF5,font-weight:bold
    classDef storage fill:#F3EEDF,stroke:#9A9488,color:#2D2A26
    subgraph DC1["主机房（上海）"]
        KP1[Kafka Primary Cluster\n3 Broker / 3 ZK]
        MM2P[MirrorMaker 2\n同步进程]
    end
    subgraph DC2["备机房（北京）"]
        KP2[Kafka Standby Cluster\n3 Broker / 3 ZK]
        MM2S[MirrorMaker 2\n反向同步进程]
    end
    APP["业务应用\n正常写主机房"]
    OPS["运维控制台\n灾备切换操作"]

    APP --> KP1
    KP1 --> MM2P --> KP2
    KP2 --> MM2S --> KP1
    OPS --> APP
    class DC1,KP1,MM2P,DC2,KP2,MM2S,APP,OPS default

MirrorMaker 2 配置

# mm2.properties - 双向同步配置
clusters = primary, standby
primary.bootstrap.servers = kafka-sh-01:9092,kafka-sh-02:9092,kafka-sh-03:9092
standby.bootstrap.servers = kafka-bj-01:9092,kafka-bj-02:9092,kafka-bj-03:9092

# 正向同步：主 -> 备
primary->standby.enabled = true
primary->standby.topics = ods\..*, dwd\..*, dws\..*
# 排除内部 Topic 和已同步的 Topic（避免循环同步）
primary->standby.topics.blacklist = .*\.internal, .*\.replica

# 反向同步：备 -> 主（用于备机房写入场景）
standby->primary.enabled = true
standby->primary.topics = ads\..*

# Offset 同步：用于恢复消费位点；是否可连续恢复还取决于位点转换、目标消费组状态和实际同步结果
primary->standby.sync.group.offsets.enabled = true
primary->standby.sync.group.offsets.interval.seconds = 60

# 性能配置
replication.factor = 3
tasks.max = 8

RPO / RTO 目标与切换演练 Checklist

指标	目标值	实现机制
RPO（恢复点目标）	< 1 分钟（设计目标，需实测）	以实际数据复制延迟及故障时未复制的数据量评估；60s 消费位点同步间隔不等于 RPO
RTO（恢复时间目标）	< 5 分钟（设计目标，需演练验证）	完成接入切换、目标 Topic 确认和有效消费位点恢复；auto.offset.reset=latest 在无有效位点时可能跳过已有消息，不能作为连续恢复保障

灾备切换演练 Checklist（每季度执行一次）：

# Step 1：确认备机房数据同步状态（Lag 应为 0 或极小值）
kafka-consumer-groups.sh --bootstrap-server kafka-bj-01:9092 \
    --group mirrormaker2-primary-to-standby --describe

# Step 2：停止主机房生产者写入（模拟主机房故障）
# [手动操作：通知各业务方暂停写入或修改配置指向备机房]

# Step 3：在源端核对实际复制消费组的进度；此检查不替代目标数据和消费位点验证
# 必须根据实际部署设置源端地址和复制消费组，不能假定示例组名存在
: "${MM2_SOURCE_BOOTSTRAP:?请设置实际源端 bootstrap.servers}"
: "${MM2_REPLICATION_GROUP:?请设置已确认的实际复制消费组}"
while true; do
    if ! output=$(kafka-consumer-groups.sh \
        --bootstrap-server "$MM2_SOURCE_BOOTSTRAP" \
        --group "$MM2_REPLICATION_GROUP" --describe); then
        echo "复制消费组查询失败，停止追平判断" >&2
        exit 1
    fi
    if ! lag=$(printf '%s\n' "$output" | awk '
        $4 ~ /^[0-9]+$/ && $5 ~ /^[0-9]+$/ && $6 ~ /^[0-9]+$/ {
            sum += $6; records++; next
        }
        $3 ~ /^[0-9]+$/ {invalid = 1}
        END {
            if (records == 0 || invalid) exit 1
            printf "%.0f\n", sum
        }'); then
        echo "无有效分区记录或位点不可用，不能判断追平" >&2
        exit 1
    fi
    echo "源端复制消费组 Lag: $lag"
    [ "$lag" -eq 0 ] && break
    sleep 10
done

# Step 4：修改业务应用的 bootstrap.servers 指向备机房
# [手动操作：更新配置中心或重启各业务应用]

# Step 5：验证备机房消费正常
kafka-consumer-groups.sh --bootstrap-server kafka-bj-01:9092 \
    --describe --all-groups | grep -v "^$" | head -30

# Step 6：记录切换完成时间，计算实际 RTO
echo "灾备切换完成，RTO = $(date)"

容量规划

消息量估算模型

容量规划的输入来自业务估算，输出是 Broker 数量、磁盘大小和网络带宽需求。

输入参数（示例：某电商平台）：
  - 每日消息总量：D = 100 亿条
  - 平均消息体大小：S = 500 字节
  - 峰值/均值流量比：P = 3 倍
  - 数据保留天数：R = 3 天
  - 副本数：RF = 3

计算过程：
  日均总数据量 = D × S = 100亿 × 500B = 5 TB/天
  日均写入速率 = 5TB / 86400s ≈ 60 MB/s
  峰值写入速率 = 60 MB/s × 3 = 180 MB/s（生产者侧）

存储容量：
  原始存储 = D × S × R = 5TB × 3 = 15 TB
  含副本存储 = 15TB × RF = 45 TB
  预留 20% 余量 = 45TB × 1.2 = 54 TB

扩容时间点判断

监控指标	扩容阈值	处理方式
Broker 磁盘使用率	> 70%	增加磁盘或迁移分区
网络带宽使用率	> 60%（峰值）	增加 Broker 节点
Consumer Lag 持续增长	Lag 增长速率 > 0 且无法追平	增加分区数或消费者实例
Broker CPU	> 70%（持续 15 分钟）	减少 Broker 上分区数或升配
ZooKeeper 请求延迟	> 100ms（p99）	优化 ZK 集群或迁移至 KRaft 模式

# 存储容量实时监控
kafka-log-dirs.sh --bootstrap-server kafka-01:9092 \
    --topic-list ods.user.behavior_event,dwd.user.behavior_detail \
    --describe 2>/dev/null | \
    python3 -c "
import sys, json
data = json.load(sys.stdin)
for broker in data['brokers']:
    total = sum(log['size'] for log_dir in broker['logDirs'] for log in log_dir['partitions'])
    print(f'Broker {broker[\"broker\"]}: {total/1024/1024/1024:.2f} GB')
"

完整全链路架构图

%%{init: {"theme":"base","themeVariables":{"fontFamily":"PingFang SC, -apple-system, Source Han Serif SC, serif","fontSize":"16px","background":"#FDFBF5","primaryColor":"#FDFBF5","primaryTextColor":"#2D2A26","primaryBorderColor":"#6E6A60","secondaryColor":"#E8EFE8","secondaryBorderColor":"#8FA890","secondaryTextColor":"#2D2A26","tertiaryColor":"#FDFBF5","tertiaryBorderColor":"#9A9488","tertiaryTextColor":"#6E6A60","lineColor":"#6E6A60","textColor":"#2D2A26","mainBkg":"#FDFBF5","nodeBorder":"#2D5A3D","clusterBkg":"#F3EEDF","clusterBorder":"#9A9488","edgeLabelBackground":"#FDFBF5"},"flowchart":{"htmlLabels":true,"curve":"basis","padding":20,"nodeSpacing":60,"rankSpacing":70,"useMaxWidth":false},"themeCSS":".node rect, .node polygon, .node circle { stroke-width: 1.5px; } .edgePath .path { stroke-width: 1.5px; } .cluster-label .nodeLabel, .cluster-label foreignObject { color: #2D2A26 !important; } .edgeLabel, .edgeLabel foreignObject { color: #2D2A26 !important; background-color: #FDFBF5; } .edgeLabel rect { fill: #FDFBF5; }"}}%%
flowchart TD
    classDef default fill:#FDFBF5,stroke:#6E6A60,color:#2D2A26,stroke-width:1.5px
    classDef startend fill:#2D2A26,stroke:#2D2A26,color:#FDFBF5,font-weight:bold
    classDef decision fill:#F6EAD8,stroke:#C08A1A,color:#2D2A26,font-weight:bold
    classDef hub fill:#3B5A80,stroke:#3B5A80,color:#FDFBF5,font-weight:bold
    classDef io fill:#E5EAF0,stroke:#3B5A80,color:#2D2A26
    classDef accent fill:#A6432C,stroke:#A6432C,color:#FDFBF5,font-weight:bold
    classDef storage fill:#F3EEDF,stroke:#9A9488,color:#2D2A26
    subgraph 数据源
        APP_SDK["APP/Web SDK\n用户行为埋点"]
        MYSQL_ORDER["MySQL 订单库\nBinlog CDC"]
        MYSQL_USER["MySQL 用户库\nBinlog CDC"]
        NGINX_LOG["Nginx 访问日志\nFilebeat 采集"]
    end

    subgraph Kafka_ODS["Kafka ODS 层"]
        ODS_BEHAVIOR["ods.user.behavior_event\n24 分区 / 7 天保留"]
        ODS_ORDER["ods.order.order_info\n12 分区 / 7 天保留"]
        ODS_USER["ods.user.user_info\n8 分区 / 7 天保留"]
        ODS_LOG["ods.infra.nginx_access\n16 分区 / 3 天保留"]
    end

    subgraph Flink_DWD["Flink DWD 作业"]
        F1["行为清洗作业\n脱敏 / 格式校验"]
        F2["订单 CDC 作业\n状态机 / 幂等去重"]
        F3["用户维度关联作业\n维表 Join"]
    end

    subgraph Kafka_DWD["Kafka DWD 层"]
        DWD_BEHAVIOR["dwd.user.behavior_detail\n24 分区 / 3 天保留"]
        DWD_ORDER["dwd.order.order_detail\n12 分区 / 3 天保留"]
    end

    subgraph Flink_ADS["Flink ADS 作业"]
        F4["实时 GMV 计算\n1 分钟窗口"]
        F5["用户漏斗分析\n会话窗口"]
        F6["订单状态聚合\n滚动窗口"]
    end

    subgraph 存储应用
        REDIS["Redis\n实时大屏"]
        HBASE["HBase\n用户画像"]
        CK["ClickHouse\n历史分析"]
    end

    APP_SDK --> ODS_BEHAVIOR
    MYSQL_ORDER --> |Canal| ODS_ORDER
    MYSQL_USER --> |Canal| ODS_USER
    NGINX_LOG --> ODS_LOG

    ODS_BEHAVIOR --> F1 --> DWD_BEHAVIOR
    ODS_ORDER --> F2 --> DWD_ORDER
    ODS_USER --> F3 --> DWD_BEHAVIOR

    DWD_BEHAVIOR --> F4 & F5
    DWD_ORDER --> F6

    F4 --> REDIS
    F5 --> HBASE
    F6 --> REDIS
    DWD_BEHAVIOR --> CK
    DWD_ORDER --> CK
    class APP_SDK,MYSQL_ORDER,MYSQL_USER,NGINX_LOG,Kafka_ODS,ODS_BEHAVIOR,ODS_ORDER,ODS_USER,ODS_LOG,Flink_DWD,F1,F2,F3,Kafka_DWD,DWD_BEHAVIOR,DWD_ORDER,Flink_ADS,F4,F5,F6,REDIS,HBASE,CK default

系统架构设计

数据采集实现（核心生产者配置）

// 高性能、高可靠生产者配置
private KafkaProducer<String, GenericRecord> createOptimizedProducer() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
            "kafka-01:9092,kafka-02:9092,kafka-03:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
    props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
            "http://schema-registry:8081");

    // 可靠性配置：ODS 层要求不丢消息
    props.put(ProducerConfig.ACKS_CONFIG, "all");
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
    props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

    // 性能配置：批量发送降低请求次数
    props.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536);        // 64KB
    props.put(ProducerConfig.LINGER_MS_CONFIG, 5);             // 最多等待 5ms 凑批
    props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
    props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);  // 64MB 发送缓冲区

    return new KafkaProducer<>(props);
}

掌握检查

完成以下检查，确保已掌握企业级 Kafka 应用：

[ ] 能描述实时数仓 ODS/DWD/ADS 三层的职责，并说明 Topic 命名规范
[ ] 理解 Canal Binlog 同步到 Kafka 的配置要点和幂等性保障方式
[ ] 能设计 Consumer Lag 告警体系，说明分级阈值的设置依据
[ ] 掌握 MirrorMaker 2 的 RPO/RTO 目标及灾备切换步骤
[ ] 能独立完成容量规划：从业务量估算出 Broker 数量和磁盘需求

练习题

某业务的 Kafka Topic 每天消息量 50 亿条，平均消息大小 1KB，保留 7 天，副本数 3，计算所需总存储容量。
Flink 作业消费 Kafka 时发现 Consumer Lag 持续增长，列举至少 5 种可能的原因和对应的排查方法。
如果订单状态机收到了乱序的 Binlog 事件（如先收到 DELIVERED 再收到 SHIPPED），如何处理？
设计一个支持多机房容灾的 Kafka 架构，要求 RPO < 30 秒，RTO < 3 分钟，画出架构图并说明关键配置。