当前位置：首页 > news >正文

Kafka

news 来源：原创 2025/7/17 19:46:43

一、什么是Kafka

核心组件

特性

使用场景

安装与配置

二、Kafka的使用

安装 ZooKeeper 和 Kafka

安装 ZooKeeper

安装 Kafka

配置 ZooKeeper 和 Kafka

配置 ZooKeeper

配置 Kafka

启动 ZooKeeper 和 Kafka

创建 Topic

编写生产者代码

编写消费者代码

运行生产者和消费者

使用 Kafka 命令行工具

三、部署Kafka一定需要部署ZooKeeper ？

KRaft 模式的优点

如何启用 KRaft 模式

注意事项

总结

一、什么是Kafka

Apache Kafka 是一个分布式流处理平台，它被设计用来处理大量的实时数据流。Kafka 由 LinkedIn 开发，并于 2011 年成为 Apache Software Foundation 的一个开源项目。它能够高效地处理大规模的消息传递，支持高吞吐量、低延迟的消息传输，同时提供了强大的持久化和容错能力。

以下是 Kafka 的一些关键概念和特性：

核心组件

Topic（主题）：
主题是 Kafka 中消息的分类或馈送名称。生产者将消息发布到特定的主题中，消费者订阅这些主题来消费消息。
Partition（分区）：
每个主题可以分为多个分区，每个分区是一个有序的、不可变的消息序列。分区允许 Kafka 在不同的节点上分布数据，从而实现水平扩展。
Broker（代理/服务器）：
Kafka 集群中的每台服务器被称为一个 broker。一个集群可以包含多个 broker，它们共同管理所有主题的数据和请求。
Producer（生产者）：
生产者是向 Kafka 发布消息的应用程序。生产者可以选择将消息发送到特定主题的哪个分区。
Consumer（消费者）：
消费者是从 Kafka 订阅消息的应用程序。消费者可以组成消费者组，这样每个消息只会被组内的一个消费者消费。
Consumer Group（消费者组）：
一个或多个消费者可以组成一个消费者组，通过组 ID 来标识。Kafka 确保每个分区的消息只会被同一个消费者组内的一个消费者消费，但不同组的消费者可以消费相同的分区。
Offset（偏移量）：
偏移量是消费者在分区中位置的一个元数据。它用于跟踪消费者已经消费到了哪条消息。Kafka 会为每个消费者组维护一个偏移量。
Replication（副本）：
分区可以有多个副本，以确保即使某个 broker 失败，数据也不会丢失。副本分布在不同的 broker 上，提供冗余和故障转移。

特性

高吞吐量：Kafka 能够处理大量消息，适用于日志聚合、监控数据处理等场景。
持久性：消息被持久化到磁盘，即使系统重启后也能保证消息不丢失。
可扩展性：Kafka 支持水平扩展，可以通过添加更多的 broker 来增加系统的容量。
容错性：通过副本机制，Kafka 提供了对 broker 故障的容忍度。
实时处理：Kafka 可以与流处理框架如 Apache Spark Streaming, Apache Flink, 和 Kafka Streams 结合使用，进行实时数据分析。

使用场景

消息队列：作为传统消息中间件的替代方案，Kafka 可以用作高性能的消息队列。
日志收集：Kafka 可以收集来自不同来源的日志数据，并将其分发给各种服务进行处理。
流处理：Kafka 适合用于构建实时数据管道和流处理应用。
事件溯源：Kafka 可以存储所有的变更事件，这对于需要完整历史记录的应用非常有用。
监控和指标：Kafka 可以收集和分发监控数据和性能指标。

安装与配置

要安装和配置 Kafka，您通常需要先安装 ZooKeeper（用于协调 Kafka 集群），然后下载并解压 Kafka 的二进制文件。接下来，您可以根据官方文档中的说明启动 ZooKeeper 和 Kafka 服务，并创建您的第一个主题。

二、Kafka的使用

安装 ZooKeeper 和 Kafka

安装 ZooKeeper

Kafka 依赖于 ZooKeeper 来管理集群元数据。您可以从 ZooKeeper 官方网站下载最新版本的 ZooKeeper，并按照官方文档进行安装。

安装 Kafka

接下来，从 Apache Kafka 官方网站下载 Kafka 的二进制文件。解压后，您会得到类似如下的目录结构：

.
├── bin
├── config
├── libs
└── ...

bin 目录包含了启动和管理 Kafka 服务的脚本，而 config 目录则包含配置文件。

配置 ZooKeeper 和 Kafka

配置 ZooKeeper

在 config/zookeeper.properties 文件中，确保指定了正确的 dataDir（存储快照的位置）和 clientPort（客户端连接端口，默认是 2181）。如果需要自定义其他设置，请参阅 ZooKeeper 文档。

配置 Kafka

在 config/server.properties 文件中，修改以下关键配置项：

broker.id：每个 broker 在集群中的唯一标识符。
listeners：指定 broker 监听的地址和端口（例如 PLAINTEXT://:9092）。
log.dirs：Kafka 日志文件的存储位置。
zookeeper.connect：指向您的 ZooKeeper 实例（例如 localhost:2181）。

启动 ZooKeeper 和 Kafka

在 Kafka 解压后的根目录下，打开两个终端窗口或命令行界面。

启动 ZooKeeper:

bin/zookeeper-server-start.sh config/zookeeper.properties

启动 Kafka:

bin/kafka-server-start.sh config/server.properties

创建 Topic

创建一个名为 test-topic 的主题，它有 3 个分区和 1 个副本因子：

bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

编写生产者代码

创建一个 Maven 项目，并在 pom.xml 中添加 Kafka 依赖：

<dependencies><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.4.0</version> <!-- 请根据实际情况选择版本 --></dependency>
</dependencies>

然后编写生产者代码，如下所示：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;public class KafkaProducerExample {public static void main(String[] args) {Properties props = new Properties();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {for (int i = 0; i < 10; i++) {String key = "key-" + i;String value = "value-" + i;ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", key, value);producer.send(record, (metadata, exception) -> {if (exception == null) {System.out.printf("Message sent to topic %s, partition %d, offset %d%n",metadata.topic(), metadata.partition(), metadata.offset());} else {exception.printStackTrace();}});}}}
}

编写消费者代码

同样，在同一个 Maven 项目中，编写消费者代码：

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class KafkaConsumerExample {public static void main(String[] args) {Properties props = new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {consumer.subscribe(Collections.singletonList("test-topic"));while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {System.out.printf("Received message: key = %s, value = %s, partition = %d, offset = %d%n",record.key(), record.value(), record.partition(), record.offset());}}}}
}

运行生产者和消费者

先启动消费者，然后运行生产者。您应该会在消费者的控制台看到由生产者发送的消息。

使用 Kafka 命令行工具

Kafka 提供了一些命令行工具来帮助管理和监控 Kafka 集群。例如，您可以使用以下命令查看主题详情、列出所有主题、删除主题等：

列出所有主题:

bin/kafka-topics.sh --list --bootstrap-server localhost:9092

描述特定主题:

bin/kafka-topics.sh --describe --topic test-topic --bootstrap-server localhost:9092

删除主题:

bin/kafka-topics.sh --delete --topic test-topic --bootstrap-server localhost:9092

使用控制台生产者和消费者:
- 控制台生产者:

bin/kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092

控制台消费者:

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning

三、部署Kafka一定需要部署ZooKeeper ？

在传统的 Kafka 部署中，ZooKeeper 是一个必需的组件，因为它负责管理 Kafka 集群的元数据、协调 broker 之间的通信以及处理故障转移。具体来说，ZooKeeper 用于：

存储集群配置：包括主题、分区和副本的信息。
维护 broker 状态：跟踪哪些 broker 是活跃的，哪些已经宕机。
消费者组管理：管理消费者组的成员关系和偏移量提交。
选举 leader：当分区的 leader 副本失效时，ZooKeeper 协助选举新的 leader。

然而，随着 Kafka 的发展，从版本 Kafka 3.0 开始引入了 KRaft 模式（Kafka Raft Metadata），这是一种不依赖于 ZooKeeper 的新元数据管理机制。KRaft 使用 Raft 共识算法来实现元数据的分布式管理和一致性，从而允许 Kafka 在没有外部依赖的情况下独立运行。

KRaft 模式的优点

简化部署：不再需要单独部署和管理 ZooKeeper 集群，减少了运维复杂度。
更好的性能：KRaft 模式下的 Kafka 可以提供更高效的元数据管理和更快的故障恢复。
更高的可用性：通过内置的 Raft 共识算法，Kafka 可以更快速地处理 broker 故障并选举新的 leader。
一致性和可靠性：KRaft 模式确保了元数据的一致性和持久性，避免了传统 ZooKeeper 模式下可能出现的一些问题。

如何启用 KRaft 模式

要使用 KRaft 模式，您需要执行以下步骤：

下载支持 KRaft 模式的 Kafka 版本：确保您使用的是 Kafka 3.0 或更高版本。
初始化 KRaft 模式：
1. 使用 kafka-storage.sh 工具初始化元数据日志。例如：
2. bash
3. 深色版本
```
bin/kafka-storage.sh format -t <cluster.id> -c config/kraft/server.properties
```
4. 这里的 <cluster.id> 是一个唯一的集群 ID，可以通过 bin/kafka-storage.sh random-uuid 生成。
修改配置文件：
1. 在 config/kraft/server.properties 中，设置 process.roles=broker,controller 和 controller.quorum.voters=<controller endpoints>。
2. 确保 listeners 和 advertised.listeners 配置正确。
启动 Kafka：
1. 使用 bin/kafka-server-start.sh 启动 Kafka，但这次是基于 KRaft 模式：
2. bash
3. 深色版本
```
bin/kafka-server-start.sh config/kraft/server.properties
```

注意事项

迁移现有集群：如果您已经有使用 ZooKeeper 的 Kafka 集群，并希望迁移到 KRaft 模式，Kafka 提供了迁移工具和支持。您可以参考官方文档中的迁移指南。
兼容性：虽然 KRaft 模式提供了许多优势，但在某些场景下，特别是对于已有系统的迁移，可能仍然需要评估是否适合立即切换到 KRaft 模式。

总结

虽然传统的 Kafka 部署确实需要 ZooKeeper，但自从 Kafka 3.0 引入 KRaft 模式后，您现在可以选择不使用 ZooKeeper 来部署 Kafka。KRaft 模式不仅简化了部署和管理，还带来了性能和可用性的提升。如果您正在规划新的 Kafka 部署，或者考虑升级现有集群，强烈建议考虑 KRaft 模式。

一、什么是Kafka

核心组件

特性

使用场景

安装与配置

二、Kafka的使用

安装 ZooKeeper 和 Kafka

安装 ZooKeeper

安装 Kafka

配置 ZooKeeper 和 Kafka

配置 ZooKeeper

配置 Kafka

启动 ZooKeeper 和 Kafka

创建 Topic

编写生产者代码

编写消费者代码

运行生产者和消费者

使用 Kafka 命令行工具

三、部署Kafka一定需要部署ZooKeeper ？

KRaft 模式的优点

如何启用 KRaft 模式

注意事项

总结

相关文章：