消息队列

Kafka知识整理

Kafka 简介

Kafka 是一个分布式的流处理平台，最初由 LinkedIn 开发，后来成为 Apache 项目。它具有高吞吐量、可靠性和可扩展性的特点，被广泛应用于日志收集、消息系统、活动追踪、流式处理等场景。

应用场景

  1. 日志聚合：收集分布式系统中的日志数据，集中存储以便分析和监控

    • 案例：电商平台将用户行为日志（浏览、点击、购买）通过Kafka收集，用于实时分析用户行为和个性化推荐

  2. 消息队列：解耦系统组件，提高系统弹性和可扩展性

    • 案例：订单系统将新订单发送到Kafka，库存系统、支付系统、物流系统各自消费消息进行处理

  3. 流处理：实时处理和转换数据流

    • 案例：金融机构使用Kafka Streams处理交易数据流，实时检测欺诈行为

  4. 事件溯源：记录状态变更事件，用于系统重建和审计

    • 案例：银行系统记录账户所有操作事件，用于账户状态重建和合规审计

  5. 指标监控：收集系统和应用指标，用于监控和告警

    • 案例：云服务提供商收集基础设施指标，实时监控系统健康状态

Kafka 架构设计

Kafka 的核心架构包含以下几个关键组件：

Broker

Broker 是 Kafka 集群中的服务器节点，负责接收和处理客户端请求，存储消息数据。每个 Broker 都有一个唯一的 ID，可以独立运行。

应用案例：大型电商平台通常部署多个Broker节点组成集群，每个节点可能处理不同类型的消息流量。例如，用户行为数据可能分配到特定的Broker组，而订单处理数据分配到另一组Broker，以实现负载均衡和资源隔离。

Producer

Producer 是消息生产者，负责将消息发送到 Kafka 集群中的特定 Topic。Producer 可以选择同步或异步的方式发送消息。

代码示例：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// 创建Producer配置
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 创建Producer实例
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 发送消息
ProducerRecord<String, String> record = new ProducerRecord<>("order-topic", orderId, orderJson);
producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        System.out.println("消息发送成功: " + metadata.offset());
    } else {
        exception.printStackTrace();
    }
});

// 关闭Producer
producer.close();

应用场景：在微服务架构中，订单服务作为Producer，将新创建的订单信息发送到Kafka，实现与库存服务、支付服务等的解耦。

Consumer

Consumer 是消息消费者，负责从 Kafka 集群中订阅并消费消息。Consumer 可以单独消费，也可以组成 Consumer Group 共同消费。

代码示例：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// 创建Consumer配置
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("group.id", "inventory-service");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");

// 创建Consumer实例
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 订阅Topic
consumer.subscribe(Arrays.asList("order-topic"));

// 消费消息
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("收到订单: " + record.key() + ", 内容: " + record.value());
        // 处理订单逻辑
        processOrder(record.value());
    }
    // 手动提交偏移量
    consumer.commitSync();
}

应用场景：库存服务作为Consumer，消费订单Topic中的消息，实时更新库存数量。多个库存服务实例可以组成Consumer Group，每个实例处理部分分区的消息，实现负载均衡。

Topic

Topic 是消息的逻辑分类，每个 Topic 可以有多个 Partition。Producer 发送消息到特定的 Topic，Consumer 从特定的 Topic 消费消息。

应用案例：电商平台可能设置多个不同的Topic：

  • user-events：用户行为事件（浏览、搜索、收藏）
  • orders：订单相关事件
  • inventory-changes：库存变更事件
  • payment-events：支付相关事件

这种分类使得不同的业务系统可以只关注与自己相关的消息流。

Partition

Partition 是 Topic 的物理分区，每个 Partition 是一个有序的、不可变的消息序列。Partition 的引入使得 Kafka 可以实现水平扩展和并行处理。

设计考量：

  • 分区数量决定了Topic的并行度，通常应该至少等于预期的Consumer数量
  • 实际案例：高流量电商平台的订单Topic可能配置32个分区，允许最多32个消费者实例并行处理订单

Segment

Segment 是 Partition 的物理存储单元，每个 Partition 由多个 Segment 组成。当 Segment 达到一定大小（默认1GB）或时间阈值时，会创建新的 Segment。

性能影响：Segment大小设置会影响：

  • 文件管理效率：较大的Segment减少文件数量
  • 数据清理效率：较小的Segment使过期数据清理更精确
  • 实际案例：日志系统可能使用较小的Segment（如256MB）以便及时清理过期日志

Log

Log 是 Kafka 中最基本的数据存储单元，每个 Partition 对应一个 Log，Log 由多个 Segment 文件组成。

存储结构：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

/kafka-logs/
  ├── topic1-0/                # topic1的第0个分区
  │   ├── 00000000000000000000.log    # 数据文件
  │   ├── 00000000000000000000.index  # 索引文件
  │   ├── 00000000000000367104.log    # 下一个segment
  │   └── 00000000000000367104.index
  ├── topic1-1/                # topic1的第1个分区
  │   └── ...
  └── topic2-0/                # topic2的第0个分区
      └── ...

ZooKeeper

ZooKeeper 用于管理和协调 Kafka 集群，存储元数据信息，如 Broker 节点、Topic 配置、消费者偏移量等。（注：新版本 Kafka 正在逐步减少对 ZooKeeper 的依赖，Kafka 2.8.0引入了KRaft模式，可以完全不依赖ZooKeeper）

ZooKeeper存储的关键信息：

  • /brokers/ids/[broker-id]：Broker节点信息
  • /brokers/topics/[topic]/partitions/[partition]/state：分区状态
  • /consumers/[group_id]/offsets/[topic]/[partition]：消费者偏移量（旧版本）

Kafka 副本同步方式

Kafka 提供了三种不同的副本同步方式，通过 acks 参数控制：

1. ack=0 (半同步复制)

  • Producer 发送消息后不等待任何确认
  • 最高的吞吐量，但无法保证消息已被接收
  • 可能导致消息丢失
  • 适用于对数据一致性要求不高的场景，如日志收集

2. ack=1 (异步复制)

  • Producer 发送消息后，等待 Leader 副本确认
  • 不等待 Follower 副本同步完成
  • 在 Leader 崩溃时可能丢失数据
  • 吞吐量和可靠性的折中方案

3. ack=all/-1 (同步复制)

  • Producer 发送消息后，等待所有 ISR 中的副本确认
  • 最高的可靠性，但吞吐量最低
  • 只要有一个 ISR 中的副本存活，就不会丢失数据
  • 适用于对数据一致性要求高的场景，如金融交易

ISR 机制 (In-Sync Replicas)

  • ISR 是与 Leader 保持同步的副本集合
  • AR (Assigned Replicas) = ISR (In-Sync Replicas) + OSR (Out-of-Sync Replicas)
  • 副本滞后超过 replica.lag.time.max.ms 会被踢出 ISR
  • 当副本重新追上 Leader 时，会被重新加入 ISR
  • ISR 机制是 Kafka 实现高可用和数据一致性的核心

Kafka 消息发送流程

  1. Producer 创建 ProducerRecord，指定 Topic 和消息内容
  2. 消息经过序列化器、分区器处理
  3. 分区器根据 Key 或轮询方式选择目标 Partition
  4. 消息被添加到内存中的批次 (Batch)
  5. Sender 线程定期将批次发送到对应的 Broker
  6. Broker 接收消息并写入对应 Partition 的 Leader 副本
  7. 根据 acks 配置，等待副本同步完成
  8. 返回响应给 Producer

Kafka 消息存储流程

  1. Broker 接收到消息后，将其追加到对应 Partition 的当前活跃 Segment 中
  2. 消息以追加写的方式写入磁盘，提高写入效率
  3. 消息按照 Offset 顺序存储，每条消息有唯一的 Offset
  4. 当 Segment 达到配置的大小或时间阈值，创建新的 Segment
  5. 旧的 Segment 会在配置的保留时间后被删除或压缩
  6. Kafka 使用页缓存和零拷贝技术优化 I/O 性能

Kafka 消息消费流程

  1. Consumer 向 Coordinator 发送 JoinGroup 请求加入消费组
  2. Coordinator 选择一个 Consumer 作为 Leader，进行分区分配
  3. 分配结果通过 SyncGroup 请求同步给所有 Consumer
  4. Consumer 向对应的 Broker 发送 Fetch 请求获取消息
  5. Broker 返回消息给 Consumer
  6. Consumer 处理消息并定期提交消费位移 (Offset)
  7. 位移提交可以是自动的或手动的，保存在内部 Topic __consumer_offsets 中

主从同步

Kafka 的主从同步基于 Leader-Follower 模型：

  1. 每个 Partition 有一个 Leader 和多个 Follower
  2. 所有读写请求都由 Leader 处理
  3. Follower 通过 Fetch 请求从 Leader 拉取消息
  4. HW (High Watermark) 表示所有 ISR 副本都已复制的位置
  5. LEO (Log End Offset) 表示每个副本的日志末端位置
  6. 消费者只能消费到 HW 位置的消息，保证数据一致性
  7. 当 Leader 失效时，从 ISR 中选举新的 Leader

高可用

Kafka 通过以下机制实现高可用：

  1. 多副本机制：每个 Partition 可以配置多个副本，分布在不同的 Broker 上
  2. Leader 选举：当 Leader 失效时，Controller 会从 ISR 中选择一个 Follower 成为新的 Leader
  3. Controller 选举：集群中的一个 Broker 会被选为 Controller，负责分区分配和故障转移
  4. Rebalance 机制：当 Consumer 加入或离开消费组时，会触发 Rebalance，重新分配分区
  5. 自动平衡：Kafka 支持自动平衡 Leader 分区，避免单个 Broker 负载过高

消息顺序

Kafka 对消息顺序的保证：

  1. 单个 Partition 内的消息是有序的
  2. 不同 Partition 之间的消息无法保证顺序
  3. 如果需要全局顺序，可以使用只有一个 Partition 的 Topic
  4. 如果需要按 Key 顺序，可以确保相同 Key 的消息路由到同一个 Partition

消息重复

消息重复的原因和处理：

  1. 原因：网络问题、Broker 崩溃、Consumer 崩溃等导致重试或重新消费
  2. Producer 端：启用幂等性 (enable.idempotence=true) 和事务功能
  3. Consumer 端：实现幂等消费，如使用唯一标识、状态检查、分布式锁等
  4. 最佳实践：设计业务逻辑时考虑幂等性，确保多次处理同一消息不会产生副作用

消息丢失

消息丢失的场景和防止措施：

  1. Producer 端：

    • 使用 acks=all 确保所有 ISR 副本都收到消息
    • 启用重试机制 (retries 参数)
    • 使用回调机制确认消息发送结果

  2. Broker 端：

    • 配置足够的副本数 (replication.factor>=3)
    • 配置最小 ISR 数量 (min.insync.replicas>=2)
    • 合理配置刷盘策略 (log.flush.* 参数)

  3. Consumer 端：

    • 手动提交位移，确保消息处理成功后再提交
    • 使用事务确保消息处理和位移提交的原子性
    • 避免长时间处理单条消息，防止会话超时

消息积压

消息积压的原因和解决方案：

  1. 原因：

    • Consumer 处理能力不足
    • 突发流量高峰
    • Consumer 异常或宕机
    • 网络问题

  2. 解决方案：

    • 增加 Consumer 实例和 Partition 数量
    • 优化 Consumer 处理逻辑，提高处理效率
    • 实现背压机制，控制生产速度
    • 使用更高性能的硬件
    • 临时将消息转储到其他存储，离线处理

消息延迟

Kafka 中的延迟消息实现：

  1. Kafka 原生不支持延迟消息，但可以通过以下方式实现：
    • 使用定时任务扫描特定 Topic
    • 使用时间轮算法在应用层实现
    • 创建多个 Topic 代表不同的延迟级别
    • 使用外部组件如 Apache Pulsar 或 RocketMQ 的延迟功能

零拷贝

Kafka 使用零拷贝技术提高性能：

  1. 传统 I/O 模型：数据在磁盘、内核空间、用户空间和网络之间多次拷贝
  2. 零拷贝技术：利用 sendfile() 系统调用，直接从磁盘到网络接口传输数据
  3. 优势：
    • 减少数据拷贝次数
    • 减少上下文切换
    • 降低 CPU 使用率
    • 提高吞吐量
  4. 应用场景：Kafka 的日志文件传输、Consumer 消费消息

Kafka 调优

Broker 调优

  • 合理设置 num.network.threads 和 num.io.threads
  • 优化 JVM 参数，如堆大小、GC 策略
  • 配置适当的 log.retention.hours 和 log.segment.bytes
  • 使用 RAID 10 磁盘阵列提高 I/O 性能

Producer 调优

  • 增大 batch.size 和 linger.ms 提高批量发送效率
  • 配置合适的 buffer.memory 避免内存溢出
  • 根据场景选择合适的 compression.type
  • 调整 max.in.flight.requests.per.connection 平衡吞吐量和顺序性

Consumer 调优

  • 合理设置 fetch.min.bytes 和 fetch.max.wait.ms
  • 优化 max.poll.records 控制单次拉取的消息数量
  • 调整 max.poll.interval.ms 避免消费者被踢出消费组
  • 实现并行处理提高消费效率

Kafka 存储结构详解

Kafka 的存储结构是其高性能的关键因素之一，它采用了分层的存储设计，从上到下依次为：Topic、Partition、Segment、Index 和 Log。

Topic 与 Partition

Topic 是消息的逻辑分类，而 Partition 是 Topic 的物理分区。每个 Topic 可以有多个 Partition，这些 Partition 分布在不同的 Broker 上，实现了数据的分布式存储和并行处理。

Partition 的数量决定了 Topic 的并行度，增加 Partition 数量可以提高吞吐量，但也会增加系统开销和复杂性。Partition 的数量一旦设定，通常不建议减少，因为这可能导致数据丢失。

Segment 文件

每个 Partition 由多个 Segment 文件组成，Segment 是 Kafka 存储的基本单位。当 Segment 达到一定大小（默认 1GB）或时间阈值时，会创建新的 Segment。

Segment 文件命名规则为：[baseOffset].[index|log|timeindex]，其中 baseOffset 是该 Segment 中第一条消息的 Offset。

每个 Segment 包含以下文件：

  • .log 文件：存储实际的消息数据
  • .index 文件：存储消息的物理位置索引
  • .timeindex 文件：存储时间戳索引（Kafka 0.10.0 版本后引入）

索引机制

Kafka 使用稀疏索引来提高查找效率。索引文件中并不是每条消息都有索引项，而是每隔一定字节数（默认 4KB）的消息才会创建一个索引项。

索引项包含两个部分：

  • 相对 Offset：消息的 Offset 相对于 Segment 基准 Offset 的值
  • 物理位置：消息在 .log 文件中的物理位置（字节偏移量）

当需要查找特定 Offset 的消息时，Kafka 首先找到该 Offset 所在的 Segment，然后在索引文件中找到小于等于目标 Offset 的最大索引项，从该位置开始顺序扫描 .log 文件，直到找到目标消息。

日志清理

Kafka 提供了两种日志清理策略：

  1. 基于时间的删除：通过 log.retention.hours 配置，删除超过保留时间的旧 Segment
  2. 基于大小的删除：通过 log.retention.bytes 配置，当 Partition 大小超过阈值时，删除最旧的 Segment
  3. 日志压缩：通过 log.cleanup.policy=compact 配置，保留每个 Key 的最新值，删除旧值

日志压缩特别适用于需要保留最新状态的场景，如配置更新、状态变更等。

文件系统与页缓存

Kafka 直接使用文件系统存储数据，而不是使用数据库。它充分利用操作系统的页缓存（Page Cache）来提高 I/O 性能：

  1. 写入操作：追加写入文件系统，由操作系统负责刷盘
  2. 读取操作：优先从页缓存读取，命中率高时可以避免磁盘 I/O

这种设计使得 Kafka 在处理大量数据时仍能保持高性能，同时简化了系统架构。

存储格式

Kafka 消息的存储格式经过精心设计，包含以下字段：

  • 8 字节 Offset
  • 4 字节消息大小
  • 4 字节 CRC32 校验和
  • 1 字节魔数（Magic Byte）
  • 1 字节属性（压缩类型等）
  • 4 字节 Key 长度（-1 表示没有 Key）
  • Key 数据（如果存在）
  • 4 字节 Value 长度
  • Value 数据（实际消息内容）

这种格式设计既保证了数据完整性，又兼顾了存储效率。

批量写入与压缩

Kafka 支持消息批量写入和压缩，以提高存储效率和网络传输效率：

  1. 批量写入：多条消息组成一个批次（Batch），一次性写入磁盘
  2. 压缩：支持 GZIP、Snappy、LZ4、ZStandard 等压缩算法
  3. 端到端压缩：Producer 压缩，Broker 保持压缩状态存储，Consumer 解压

压缩率取决于消息内容的特性，对于文本类数据，通常可以达到 3-5 倍的压缩比。

存储优化最佳实践

  1. 合理设置 Partition 数量：

    • 考虑并行度需求和资源限制
    • 一般建议每个 Broker 的 Partition 数不超过 2000-4000

  2. 优化磁盘配置：

    • 使用 SSD 提高随机读写性能
    • 使用 RAID 10 而非 RAID 5/6
    • 分离操作系统和数据目录

  3. 调整 Segment 大小：

    • 较小的 Segment 有利于及时清理过期数据
    • 较大的 Segment 减少文件数量，降低管理开销

  4. 合理配置保留策略：

    • 根据业务需求设置 log.retention.hours 和 log.retention.bytes
    • 对不同 Topic 设置不同的保留策略

  5. 监控磁盘使用率：

    • 保持足够的磁盘空间（至少 20% 空闲）
    • 设置磁盘使用率告警

总结

Kafka 作为一个高性能、分布式的流处理平台，通过精心设计的架构和机制，实现了高吞吐量、可靠性和可扩展性。理解 Kafka 的核心概念和工作原理，对于构建高效、可靠的消息系统至关重要。

Licensed under CC BY-NC-SA 4.0
© 2024 - 2025 Hollis Ho
使用 Hugo 构建
主题 Stack 由 Jimmy 设计