2. Kafka API 概览：五大核心 API 的功能与适用场景

Apache Kafka 提供了五组核心 API，分别覆盖消息生产、消息消费、集群管理、流式处理与数据集成。理解每一组 API 的定位、能力边界与典型场景，是进行后续契约设计与诊断校验的基础。

2.1 五大 API 全景

下表从功能定位、核心接口、适用场景三个维度对五大 API 进行横向对比：

2.2 Producer API：消息生产契约

Producer API 是 Kafka 数据流的入口。应用程序通过 KafkaProducer 实例将 ProducerRecord 发送到指定主题，支持同步发送、异步回调与批量发送三种模式。

• 核心参数： bootstrap.servers、key.serializer、value.serializer、acks、retries
• 发送模式： fire-and-forget、同步 send().get()、异步 send() + Callback
• 分区策略： 默认轮询、key 哈希、自定义 Partitioner
• 适用场景： 埋点日志、业务事件、监控指标、异步通知

// Producer API 基础示例
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("orders", "order-1001", "{\"userId\":42,\"amount\":199.0}"),
    (metadata, exception) -> {
        if (exception != null) exception.printStackTrace();
        else System.out.println("Sent to partition " + metadata.partition());
    });
producer.close();

2.3 Consumer API：消息消费契约

Consumer API 负责从主题中拉取消息，支持消费者组实现负载均衡与故障转移。消费位移（offset）的提交策略直接影响消息的 at-least-once 或 exactly-once 语义。

• 核心参数： bootstrap.servers、key.deserializer、value.deserializer、group.id、enable.auto.commit
• 消费模式： 自动提交位移、手动同步提交、手动异步提交
• 再均衡监听器： ConsumerRebalanceListener 用于处理分区分配变更
• 适用场景： 实时流处理、消息队列、数据管道消费者

// Consumer API 基础示例
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "order-processor");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("orders"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n",
            record.offset(), record.key(), record.value());
    }
    consumer.commitSync(); // 手动提交位移
}

2.4 AdminClient API：集群管理契约

AdminClient API 提供了对 Kafka 集群的编程式管理能力，无需依赖命令行工具或 Kafka Manager。在诊断契约校验中，AdminClient 常用于校验集群元数据是否符合预期。

• 核心操作： 创建/删除主题、列出主题、描述集群、修改配置、查看消费者组
• 异步设计： 所有操作返回 Future 或 KafkaFuture，支持链式调用
• 适用场景： 自动化运维、CI/CD 环境准备、契约校验前置检查

// AdminClient API 基础示例
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");

try (AdminClient admin = AdminClient.create(props)) {
    // 创建主题
    NewTopic newTopic = new NewTopic("orders-events", 6, (short) 3);
    CreateTopicsResult result = admin.createTopics(Arrays.asList(newTopic));
    result.all().get(); // 等待完成

    // 描述主题
    DescribeTopicsResult desc = admin.describeTopics(Arrays.asList("orders-events"));
    TopicDescription topicDesc = desc.values().get("orders-events").get();
    System.out.println("Partitions: " + topicDesc.partitions().size());
}

2.5 Streams API：有状态流处理

Streams API 构建在 Producer API 和 Consumer API 之上，提供声明式流处理 DSL。它支持事件时间窗口、状态存储（RocksDB）、KTable-KStream 连接等高级语义，适合构建实时 ETL 与微服务。

• 核心抽象： KStream（无界记录流）、KTable（变更日志表）、GlobalKTable（全局表）
• 状态操作： groupBy + aggregate、join、windowedBy
• 适用场景： 实时聚合、异常检测、数据清洗、事件驱动微服务

// Streams API 基础示例（实时订单金额聚合）
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> orders = builder.stream("orders");
KGroupedStream<String, Long> grouped = orders
    .mapValues(value -> Long.parseLong(JsonPath.read(value, "$.amount").toString()))
    .groupByKey();
KTable<Windowed<String>, Long> windowedCounts = grouped
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .reduce(Long::sum);
windowedCounts.toStream().to("orders-amount-5min");

2.6 Connect API：可扩展数据集成

Connect API 提供标准化的连接器框架，用于将 Kafka 与外部系统（数据库、文件系统、搜索引擎）集成。Source Connector 负责导入数据，Sink Connector 负责导出数据。

• 核心概念： Connector（连接器）、Task（任务）、Converter（数据转换器）、Transform（单条转换）
• 运行模式： 单机模式（Standalone）、分布式模式（Distributed）
• 适用场景： MySQL CDC、Elasticsearch 同步、HDFS 导出、日志文件采集

2.7 五大 API 协作关系

下图展示了五大 API 在 Kafka 生态中的协作关系与数据流向：

2.8 API 选型决策要点

在实际项目中，选择哪一组 API 取决于业务需求与架构约束。以下决策树可辅助判断：

1. 是否需要有状态处理？ 是 → Streams API；否 → 进入下一步
2. 是否需要与外部系统集成？ 是 → Connect API；否 → 进入下一步
3. 是否需要管理集群元数据？ 是 → AdminClient API；否 → 进入下一步
4. 仅需生产/消费消息？ 使用 Producer API + Consumer API