Kafka API

2025年1月23日 | 阅读 9 分钟

引言

Apache Kafka 最初由 LinkedIn 开发，现已成为 Apache 软件基金会的骄傲之作，它是一个用 Scala 和 Java 编写的高性能分布式流处理平台。Kafka 在处理实时数据流方面表现出色，吞吐量高且延迟低。Kafka API 对于与 Kafka 集群交互至关重要，它使开发人员能够利用 Kafka 的能力来构建可伸缩且容错的应用程序。本综合指南将详细介绍 Kafka API，并提供实用的示例和代码片段，使学到的知识引人入胜且有效。

什么是 Kafka？

Kafka 是一个多功能平台，具有三个主要功能：

发布和订阅记录流：Kafka 允许应用程序发布（写入）和订阅（读取）数据流。
以容错方式存储记录流：Kafka 的分布式提交日志可靠地存储数据，确保数据的持久性和容错性。

三、处理记录流：Kafka 允许实时处理数据流，使应用程序能够对到达的数据做出响应。

Kafka 架构

在深入研究 Kafka API 之前，了解 Kafka 的架构至关重要。Kafka 的架构包含几个关键组件：

生产者（Producer）： 将记录发送到 Kafka 主题的应用程序。
消费者（Consumer）： 从 Kafka 主题读取数据的应用程序。
代理（Broker）： 存储数据并为客户端提供服务的 Kafka 服务器。
主题（Topic）： 发送数据的类别，也是读取数据的来源。
分区（Partition）： 主题被划分为分区，以实现可伸缩性和并行处理。
ZooKeeper： 管理和协调 Kafka 代理的服务。

Kafka API 概述

Kafka API 分为多个核心组件，每个组件都有其特定用途。这些组件包括：

生产者 API（Producer API）： 允许应用程序将数据流发送到 Kafka 集群中的主题。
消费者 API（Consumer API）： 允许应用程序从 Kafka 集群中的主题读取数据流。
Streams API： 允许应用程序处理数据流。
Connect API： 此 API 使构建和运行可重用的数据导入/导出连接器成为可能，这些连接器可以在 Kafka 和其他系统之间传输数据。

生产者 API

生产者 API 允许应用程序将数据发送到 Kafka 主题。生产者将数据推送到主题，然后由 Kafka 代理进行处理。这个过程对于 Kafka 作为高吞吐量消息代理的作用至关重要。

生产者 API 中的关键概念

ProducerRecord： 表示正在发送到主题的记录。它包含主题名称、分区号、键、值和时间戳。

序列化器（Serializer）： 将键和值对象转换为字节数组。Kafka 提供默认序列化器，您也可以实现自定义序列化器。

ProducerConfig： 生产者的配置，包括批处理、压缩、重试和确认设置。

示例：创建 Kafka 生产者

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
public class SimpleProducer {
    public static void main(String[] args) {
        String topicName = "exampleTopic";
        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");
        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>(topicName, Integer.toString(i), "Message " + i);
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) {
                        System.out.printf("Sent record to topic %s partition %d with offset %d%n",
                                          metadata.topic(), metadata.partition(), metadata.offset());
                    } else {
                        exception.printStackTrace();
                    }
                }
            });
        }
        producer.close();
    }
}

在此示例中，生产者将十条消息发送到名为“exampleTopic”的主题。`send` 方法是异步的，并允许使用回调来处理发送操作的结果、提供元数据或处理异常。

消费者 API

消费者 API 允许应用程序从 Kafka 主题读取数据。消费者订阅主题并处理到达的数据，从而实现实时数据处理和分析。

消费者 API 中的关键概念

ConsumerRecord： 表示从主题获取的记录。它包含元数据，例如偏移量、分区、键和值。
反序列化器（Deserializer）： 将返回的字节数组转换为对象。Kafka 提供默认反序列化器，您也可以实现自定义反序列化器。
ConsumerConfig： 消费者的配置，包括组管理、心跳、轮询间隔和偏移量管理设置。

示例：创建 Kafka 消费者

以下是一个简单的 Java Kafka 消费者示例：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Collections;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) {
        String topicName = "exampleTopic";
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(topicName));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

在此示例中，消费者订阅名为“exampleTopic”的主题，并不断轮询新数据。`poll` 方法从 Kafka 检索数据，并打印每条记录的偏移量、键和值。

Streams API

Streams API 允许应用程序实时处理数据。它构建在核心 Kafka 生产者和客户端库之上，为流处理提供了更高级别的抽象，允许执行过滤、连接和聚合数据流等复杂操作。

Streams API 中的关键概念

KStream： 表示数据流。它是处理流中数据的抽象。
KTable： 表示更新的变更日志流。它是处理有状态操作的抽象。
Processor： 处理拓扑中的一个节点。它代表核心处理逻辑。

示例：创建流处理应用程序

以下是一个统计消息中单词计数的 Kafka Streams 应用程序示例：

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.StreamsConfig;
import java.util.Arrays;
import java.util.Properties;

public class StreamProcessingExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-example");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream("input-topic");
        KTable<String, Long> wordCounts = source
            .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
            .groupBy((key, word) -> word)
            .count();

        wordCounts.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}

在此示例中，应用程序从“input-topic”读取消息，将消息值拆分为单词，计算每个单词的出现次数，并将单词计数写入“output-topic”。

Connect API

Connect API 简化了 Kafka 与其他系统（包括数据库和文件系统）的集成。它为常见的数据源和汇点提供了可重用的连接器，从而无需编写自定义集成代码即可更轻松地将数据移入和移出 Kafka。

Connect API 中的关键概念

Connector： 用于从源提取数据或将数据推送到汇点的插件。连接器可以是源连接器（将数据上传到 Kafka）或汇点连接器（将数据导出到 Kafka）。
Source Connector： 将数据从外部系统导入 Kafka。它从源系统读取数据并将其发送到 Kafka 主题。
Sink Connector： 将数据从 Kafka 导出到外部系统。它从 Kafka 主题读取数据并将其写入汇点系统。

示例：设置 Kafka Connect 源连接器

以下是一个 JDBC 源连接器的示例配置，该连接器将数据从 MySQL 数据库导入 Kafka：

{
  "name": "jdbc-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:mysql://:3306/mydb",
    "

connection.user": "user",
    "connection.password": "password",
    "topic.prefix": "jdbc-",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "table.whitelist": "my_table"
  }
}

在此配置中，JDBC 源连接器连接到 MySQL 数据库，从“my_table”表读取数据，并将数据写入以“jdbc-”为前缀的 Kafka 主题。

Kafka API 的实际应用

Kafka API 可用于多种实际场景。以下是一些实际应用：

实时分析： Kafka 可以收集实时分析数据并实时处理，从而实现即时洞察和操作。
数据集成： Kafka Connect 可以在不同的数据系统之间实现无缝集成，包括数据库、数据仓库和云服务。
事件溯源： Kafka 的持久化日志非常适合事件溯源架构，其中状态更改被记录为事件序列。
流处理： Kafka Streams 提供了强大的工具来实时处理数据流，包括过滤、聚合和连接流。

处理 Kafka API 错误

处理 Kafka 中的错误对于构建健壮的应用程序至关重要。以下是一些常见的错误处理策略：

重试： 为瞬时错误配置重试。Kafka 允许您配置重试次数和退避策略。

死信队列（Dead Letter Queues）： 将失败的数据发送到单独的主题以供以后分析。这有助于隔离复杂的数据。

日志记录： 详细记录错误以进行故障排除。适当的日志记录有助于识别和诊断问题。

示例：Kafka 生产者中的错误处理

以下是一个带有错误处理的 Kafka 生产者的示例：

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class ErrorHandlingProducer {
    public static void main(String[] args) {
        String topicName = "exampleTopic";
        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("retries", 3);
        props.put("acks", "all");

        Producer<String, String> producer = new KafkaProducer<>(props);

        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>(topicName, Integer.toString(i), "Message " + i);
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("Error producing message: " + exception.getMessage());
                    } else {
                        System.out.printf("Produced message to topic %s partition %d offset %d%n",
                                metadata.topic(), metadata.partition(), metadata.offset());
                    }
                }
            });
        }
        producer.close();
    }
}

在此示例中，生产者配置为在发生故障时最多重试发送消息三次。回调处理每次发送操作的结果，记录错误或成功消息。

高级 Kafka API 功能

1. Kafka 事务

Kafka 支持事务，以确保一系列操作要么全部成功，要么全部失败。这对于保持数据一致性至关重要，尤其是在需要原子地生产或消费多个数据片段的情况下。

示例：Kafka 事务性生产者

以下是一个 Kafka 事务性生产者的示例：

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class TransactionalProducer {
    public static void main(String[] args) {
        String topicName = "exampleTopic";
        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("transactional.id", "transactional-producer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        producer.initTransactions();

        try {
            producer.beginTransaction();
            for (int i = 0; i < 10; i++) {
                ProducerRecord<String, String> record = new ProducerRecord<>(topicName, Integer.toString(i), "Message " + i);
                producer.send(record);
            }
            producer.commitTransaction();
        } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
            producer.close();
        } catch (KafkaException e) {
            producer.abortTransaction();
        }
        producer.close();
    }
}

在此示例中，生产者启动事务，发送一系列数据，然后提交事务。如果发生错误，事务将被中止以确保数据一致性。

Kafka Streams 交互式查询

Kafka Streams 提供了一个交互式查询功能，允许应用程序查询流处理应用程序的状态。这对于检索 KTable 或 KStream 的当前状态非常有用。

示例：交互式查询

以下是一个带有交互式查询的 Kafka Streams 应用程序示例：

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.apache.kafka.streams.StreamsConfig;

import java.util.Properties;

public class InteractiveQueriesExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-example");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream("input-topic");
        KTable<String, Long> wordCounts = source
            .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
            .groupBy((key, word) -> word)
            .count();

        wordCounts.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        // Accessing the state store
        ReadOnlyKeyValueStore<String, Long> keyValueStore =
            streams.store("Counts", QueryableStoreTypes.keyValueStore());
        System.out.println("Count for 'example': " + keyValueStore.get("example"));
    }
}

在此示例中，应用程序构建了一个统计单词的流处理拓扑，并将计数存储在 KTable 中。通过交互式查询状态存储来检索特定单词的计数。

Kafka Connect 自定义连接器

创建自定义 Kafka 连接器允许您将 Kafka 与特定或专有数据系统集成。可以为源操作和汇点操作构建自定义连接器，从而扩展 Kafka 的集成功能。

示例：自定义 Kafka 连接器

以下是一个 Java 自定义源连接器任务的示例：

import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;

import java.util.Collections;
import java.util.List;
import java.util.Map;

public class CustomSourceTask extends SourceTask {
    @Override
    public String version() {
        return "1.0";
    }
    public void start(Map<String, String> props) {
        // Initialize your task
    }
    public List<SourceRecord> poll() throws InterruptedException {
        // Poll for new records from the data source
        SourceRecord record = new SourceRecord(
            Collections.singletonMap("sourcePartition", "partition1"),
            Collections.singletonMap("sourceOffset", 42),
            "targetTopic",
            null,
            null,
            null,
            "value"
        );
        return Collections.singletonList(record);
    }

    @Override
    public void stop() {
        // Clean up resources
    }
}

在此示例中，`CustomSourceTask` 类扩展了 `SourceTask` 并实现了从源轮询数据并创建发送到 Kafka 的 `SourceRecord` 实例的关键方法。

结论

Kafka 强大的 API 使开发人员能够构建可伸缩、容错且实时的**数据**处理应用程序。通过理解和利用生产者、消费者、Streams 和 Connect API，您可以充分发挥 Kafka 的潜力。本综合指南为您提供了开始使用 Kafka API 的坚实基础，其中包含示例和实际应用。立即深入 Kafka 生态系统，开始构建强大的数据驱动型应用程序。无论您是从事实时分析、数据集成、事件溯源还是流处理，Kafka 都为您提供了满足需求的工具和能力。

下一个主题Kafka 命令行工具

Kafka API

引言

什么是 Kafka？

Kafka 架构

Kafka API 概述

生产者 API

消费者 API

Streams API

Connect API

Kafka API 的实际应用

处理 Kafka API 错误

示例：Kafka 生产者中的错误处理

高级 Kafka API 功能

1. Kafka 事务

Kafka Streams 交互式查询

Kafka Connect 自定义连接器

结论

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Kafka 教程

Kafka 安装

Kafka CLI

Kafka 编程

实时示例

Kafka 监控

Kafka Connect

Kafka Streams

杂项

Kafka API

引言

什么是 Kafka？

Kafka 架构

Kafka API 概述

生产者 API

消费者 API

Streams API

Connect API

Kafka API 的实际应用

处理 Kafka API 错误

示例：Kafka 生产者中的错误处理

高级 Kafka API 功能

1. Kafka 事务

Kafka Streams 交互式查询

Kafka Connect 自定义连接器

结论

相关帖子

使用 Kafka 进行事件驱动的供应链优化

Kafka Streams DSL

Kafka 安全性

Kafka 镜像

Kafka 超越发布-订阅：高级用法模式

节能 Kafka 集群

消息传递语义

使用 Kafka 的预测性缓存用例

Kafka 延迟

使用 Kafka 主题流和 Cosmos DB 进行数据丰富

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器