kafka-development

Kafka 开发

本技能介绍 Apache Kafka 事件流和分布式消息系统的最佳实践。在构建基于 Kafka 的应用程序时，请遵循以下指南。

核心原则

• Kafka 是一个分布式事件流平台，用于高吞吐量、容错消息传递
• 与传统的发布/订阅模式不同，Kafka 使用拉取模型：消费者从分区拉取消息
• 根据需要，围绕可扩展性、持久性和精确一次语义进行设计
• 实现中不得留下任何待办事项、占位符或缺失部分

工作流：搭建 Kafka 生产者-消费者管道

1. 定义主题 — 选择一个描述性名称，根据预期的消费者并行度设置分区数，并配置保留时间和复制因子。
2. 设计消息架构 — 在 Schema Registry 中注册 Avro 或 JSON 架构；从一开始就确保向后兼容。
3. 实现生产者 — 配置 acks=all，启用幂等性，选择分布均匀的分区键，并添加带重试逻辑的错误处理。
4. 实现消费者 — 设置 enable.auto.commit=false，选择合适的 auto.offset.reset 策略，以幂等方式处理消息，并且仅在处理成功后提交偏移量。
5. 添加可观测性 — 监控生产者发送速率、消费者滞后量和代理未充分复制的分区；在消息头中传递跟踪上下文。
6. 端到端测试 — 使用 Testcontainers 或嵌入式 Kafka 代理验证完整的生产、消费和提交周期，包括故障和再平衡场景。
7. 部署和监控 — 部署时配置滞后告警、将持续失败的消息路由到死信主题，并为关键代理和客户端指标配置仪表板。

架构概览

核心组件

• 主题：用于组织消息的类别或消息流
• 分区：主题内有序且不可变的序列，可实现并行处理
• 生产者：向主题发布消息的客户端
• 消费者：从主题读取消息的客户端
• 消费者组：协调多个消费者共同消费
• 代理：存储数据并为客户端提供服务的 Kafka 服务端

关键概念

• 消息按顺序追加到分区中
• 每条消息都有一个偏移量，即分区内唯一的顺序 ID
• 消费者维护自己的游标（偏移量），并可重复读取消息流
• 分区分布在多个代理上，以实现可扩展性

主题设计

分区策略

• 使用分区键，将相关事件放入同一分区
• 键相同的消息始终进入同一分区
• 这可确保相关事件的顺序
• 谨慎选择键：分布不均会导致分区过热

分区数

• 分区越多，并行度越高，但开销也越大
• 请考虑预期吞吐量、消费者数量和代理资源
• 初始分区数可设为预期并发运行的消费者数量
• 可以增加分区，但不能减少

主题配置

• retention.ms：消息保留时长（默认 7 天）
• retention.bytes：每个分区的最大容量
• cleanup.policy：delete（删除旧消息）或 compact（保留每个键的最新消息）
• min.insync.replicas：必须确认的最少副本数

生产者最佳实践

可靠性设置

acks=all               # Wait for all replicas to acknowledge
retries=MAX_INT        # Retry on transient failures
enable.idempotence=true # Prevent duplicate messages on retry

性能调优

• batch.size：发送前累积的消息量（默认 16KB）
• linger.ms：批处理等待时间（0 = 立即发送）
• buffer.memory：用于缓冲待发送消息的总内存
• compression.type：使用 gzip、snappy、lz4 或 zstd 节省带宽

错误处理

• 实现带指数退避的重试逻辑
• 对可重试和不可重试异常采用不同的处理方式
• 记录发送失败日志并触发告警
• 对反复失败的消息，可考虑使用死信主题

示例：具备幂等性和错误处理的 Java 生产者

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
    ProducerRecord<String, String> record =
        new ProducerRecord<>("orders", "order-123", "{\"item\":\"widget\",\"qty\":5}");

    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            log.error("Send failed for key=order-123", exception);
            // Route to dead-letter topic or alert
        } else {
            log.info("Delivered to {}-{} offset {}",
                metadata.topic(), metadata.partition(), metadata.offset());
        }
    });
}

分区器

• 默认：根据键的哈希值确定分区（键为 null 时采用轮询）
• 针对特定路由需求使用自定义分区器
• 确保均匀分布，避免出现热点分区

消费者最佳实践

偏移量管理

• 消费者通过偏移量跟踪已处理的消息
• auto.offset.reset：earliest（从头开始）或 latest（仅处理新消息）
• 成功处理后再提交偏移量，不要提前提交
• 使用 enable.auto.commit=false 实现精确一次语义

消费者组

• 组内的消费者共享分区（每个分区分配给一个消费者）
• 消费者数量多于分区数量时，部分消费者会处于空闲状态
• 消费者加入或离开时会触发组再平衡
• 使用 group.instance.id 实现静态成员身份，以减少再平衡

处理模式

• 按顺序处理分区内的消息
• 必要时处理跨分区的乱序消息
• 实现幂等处理，以支持至少一次投递
• 考虑采用事务处理，以实现精确一次

示例：使用手动偏移量提交的 Java 消费者

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Collections.singletonList("orders"));

    while (running) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
        for (ConsumerRecord<String, String> record : records) {
            try {
                processOrder(record.key(), record.value());
            } catch (Exception e) {
                log.error("Failed to process offset={} key={}", record.offset(), record.key(), e);
                publishToDeadLetterTopic(record, e);
            }
        }
        consumer.commitSync(); // Commit only after successful processing
    }
}

超时与故障

• 实现处理超时，以隔离处理缓慢的事件
• 超时发生时，将事件暂存一旁并继续处理下一条消息
• 相比逐一处理所有事件，优先维持系统整体性能
• 对多次重试后仍失败的消息使用死信队列

错误处理与重试

重试策略

• 每次处理尝试允许多次运行时重试
• 示例：每次重新投递时重试 3 次，最多重新投递 5 次 = 总计重试 15 次
• 运行时重试通常可处理 99% 的故障
• 重试次数用尽后，将消息路由到死信队列

死信主题

• 为无法处理的消息创建专用 DLT
• 包含原始主题、分区、偏移量和错误详情
• 监控 DLT，以发现表明系统性问题的模式
• 实现从 DLT 手动或自动重试

架构管理

架构注册表

• 使用 Confluent Schema Registry 管理架构
• 生产者在序列化期间根据已注册的架构验证数据
• 架构不匹配时会抛出异常，防止产生格式错误的数据
• 为生产者和消费者提供通用参考

架构演进

• 设计具备前向和后向兼容性的架构
• 添加带默认值的可选字段，以保持后向兼容性
• 避免删除或重命名字段
• 使用架构版本控制和迁移策略

Kafka Streams

状态管理

• 实施日志压缩，以维护每个键的最新版本
• 定期清除状态存储中的旧数据
• 监控状态存储的大小和访问模式
• 根据规模选择合适的存储后端

窗口操作

• 处理乱序事件和时间戳偏斜
• 使用适当的时间提取和水位线技术
• 为迟到数据配置宽限期
• 根据使用场景选择窗口类型（滚动、跳跃、滑动、会话）

安全

身份验证

• 使用 SASL/SSL 进行客户端身份验证
• 支持 SASL 机制：PLAIN、SCRAM、OAUTHBEARER、GSSAPI
• 启用 SSL 以加密传输中的数据
• 定期轮换凭据

授权

• 使用 Kafka ACL 实现精细化访问控制
• 为每个主体授予最低限度的必要权限
• 按主题分离读写权限
• 定期审计访问模式

监控与可观测性

关键指标

• 生产者：record-send-rate、record-error-rate、batch-size-avg
• 消费者：records-consumed-rate、records-lag、commit-latency
• 代理：under-replicated-partitions、request-latency、disk-usage

滞后监控

• 消费者滞后 = 最后生成的偏移量 - 最后提交的偏移量
• 滞后过高表明消费者无法跟上
• 对滞后持续增加的趋势发出告警
• 扩展消费者或优化处理

分布式追踪

• 在消息头中传播追踪上下文
• 使用 OpenTelemetry 进行端到端追踪
• 关联生产者和消费者的跨度
• 跟踪消息在管道中的传递过程

测试

单元测试

• 模拟 Kafka 客户端以进行隔离测试
• 测试序列化和反序列化逻辑
• 验证分区逻辑
• 测试错误处理路径

集成测试

• 使用嵌入式 Kafka 或 Testcontainers
• 测试完整的生产者-消费者流程
• 验证是否使用了精确一次语义
• 测试再均衡场景

性能测试

• 使用接近生产环境的消息速率进行负载测试
• 测试消费者吞吐量和滞后情况
• 验证负载下的代理资源使用情况
• 测试故障和恢复场景

常见模式

事件溯源

• 将所有状态变更存储为不可变事件
• 通过重放事件重建状态
• 使用日志压缩生成快照
• 支持时光回溯调试

CQRS（命令查询职责分离）

• 分离写入（命令）模型和读取（查询）模型
• 使用 Kafka 作为事件存储
• 基于事件构建针对读取优化的投影
• 妥善处理最终一致性

Saga 模式

• 协调跨服务的分布式事务
• 每个服务都发布事件以推进下一步
• 实现补偿事务以进行回滚
• 使用关联 ID 跟踪 Saga 实例

变更数据捕获（CDC）

• 将数据库变更捕获为 Kafka 事件
• 使用 Debezium 或类似的 CDC 工具
• 实现实时数据同步
• 构建事件驱动集成