从文字到使用，一文读懂Kafka服务使用

🏆作者简介，普修罗双战士，一直追求不断学习和成长，在技术的道路上持续探索和实践。
🏆多年互联网行业从业经验，历任核心研发工程师，项目技术负责人。
🎉欢迎 👍点赞✍评论⭐收藏

Kafka知识专栏学习

Kafka知识云集	访问地址	备注
Kafka知识点(1)	https://blog.csdn.net/m0_50308467/article/details/134138971	Kafka专栏
Kafka知识点(2)	https://blog.csdn.net/m0_50308467/article/details/134733596	Kafka专栏

🔎 Kafka 入门学习

🍁01. 什么是Kafka？

Kafka是一个分布式流处理平台，最初由LinkedIn开发并开源。它被设计用于处理大规模的实时数据流，并能够持久化和发布数据记录。Kafka具有高吞吐量、可扩展性和容错性的特点，使得它成为构建实时流处理应用程序的理想选择。

Kafka的核心概念包括以下几个要素：

1. 主题（Topic）：主题是Kafka中的数据流分类，类似于消息队列中的队列。数据被发布到不同的主题中，消费者可以订阅感兴趣的主题来消费数据。

2. 分区（Partition）：主题可以被分为多个分区，每个分区是一个有序的日志文件，用于存储消息。分区允许数据在集群中并行处理和存储，提高了吞吐量和可扩展性。

3. 副本（Replication）：每个分区可以有多个副本，副本用于提供数据的冗余备份和容错性。副本分布在不同的服务器上，当某个副本不可用时，可以从其他副本中获取数据。

4. 生产者（Producer）：生产者是数据的发送者，负责将数据发布到Kafka的主题中。生产者可以将数据发送到指定的分区，也可以让Kafka根据一定的策略自动选择分区。

5. 消费者（Consumer）：消费者是数据的接收者，负责从Kafka的主题中读取数据。消费者可以以不同的方式消费数据，如按照时间顺序、按照分区等。

Kafka的应用场景非常广泛，包括日志收集、事件驱动架构、实时分析、流式处理等。它可以处理大规模的数据流，并具有高吞吐量、可靠性和可扩展性的特点，因此在处理实时数据和构建大规模数据流处理系统方面非常有价值。

🍁02. Kafka有什么特点，优缺点是什么？

Kafka具有以下特点：

1. 高吞吐量：Kafka能够处理大规模的数据流，并提供很高的吞吐量。它通过将数据分区和并行处理，以及支持批量发送和接收，实现了高效的数据处理。

2. 可扩展性：Kafka的分布式架构使得它可以轻松地扩展到多个服务器上，以适应不断增长的数据流量和负载。通过增加分区和副本，Kafka可以水平扩展，提高系统的容量和性能。

3. 持久性：Kafka将数据持久化到磁盘中，以确保数据的可靠性和持久性。数据存储在分区中，并且可以根据需要保留一段时间，以便后续的数据分析和处理。

4. 实时处理：Kafka是一个实时流处理平台，能够处理实时数据流。它支持低延迟的数据传输和处理，可以实时地将数据从生产者传递给消费者，并支持流式处理和实时分析。

5. 容错性：Kafka通过将数据复制到多个副本中来提供容错性。当某个副本不可用时，可以从其他副本中获取数据，确保数据的可靠性和可用性。

6. 灵活性：Kafka具有灵活的消息传递模型，可以根据需要进行消息的发布和订阅。它支持多种消息格式和协议，并提供了丰富的API和工具，使得开发者可以根据自己的需求进行定制和扩展。

Kafka的优点包括高吞吐量、可扩展性、持久性和实时处理能力，使得它成为处理大规模实时数据流的理想选择。然而，Kafka也有一些缺点：

1. 复杂性：Kafka的配置和管理相对复杂，需要一定的专业知识和经验。对于初学者来说，上手可能会有一定的学习曲线。

2. 存储成本：由于Kafka将数据持久化到磁盘中，需要一定的存储空间。对于大规模的数据流，存储成本可能会成为一个考虑因素。

3. 延迟：尽管Kafka具有较低的延迟，但在某些场景下，特别是对于需要严格实时性的应用程序来说，延迟可能仍然是一个限制因素。

总体而言，Kafka是一个功能强大的分布式流处理平台，具有高吞吐量、可扩展性和持久性的优点，适用于处理大规模实时数据流的应用场景。然而，对于一些特定的需求，如低延迟和简单性，可能需要进行权衡和评估。

🍁03. Kafka 与 Zookeeper 服务的关系？

Kafka与Zookeeper是两个独立但密切相关的组件，它们在Kafka集群中扮演不同的角色。

Zookeeper是一个分布式的协调服务，用于管理和维护Kafka集群的元数据信息，包括主题、分区、消费者组等。Kafka使用Zookeeper来进行以下任务：

1. 集群管理：Kafka集群中的每个节点都在Zookeeper中注册自己的信息，包括节点的IP地址、端口号等。Zookeeper负责监控和管理这些节点，以确保集群中的节点状态的一致性。

2. Leader选举：Kafka的每个分区都有一个分区领导者（Leader），负责处理该分区的读写请求。如果分区领导者不可用，Zookeeper会协助进行新的领导者选举，选择一个新的分区领导者。

3. 消费者组协调：Kafka中的消费者可以组成消费者组，共同消费一个主题。Zookeeper负责协调消费者组的工作，包括分配分区给消费者、监控消费者的健康状态等。

4. 动态扩缩容：当Kafka集群需要扩展或缩小规模时，Zookeeper可以帮助协调新节点的加入和旧节点的离开，以实现集群的动态扩缩容。

总结来说，Zookeeper在Kafka集群中扮演着重要的角色，用于管理和维护集群的元数据信息、协调分区的领导者选举以及消费者组的协调工作。Kafka依赖于Zookeeper来实现这些功能，因此在使用Kafka时需要同时启动和管理Zookeeper服务。

🍁04. Kafka 的常见命令有哪些？

Kafka的常见命令包括：

1. kafka-topics.sh：用于创建、查看和管理主题的命令。

   • 创建主题： kafka-topics.sh --create --topic my-topic --partitions 3 --replication-factor 1 --bootstrap-server localhost:9092
   • 查看主题列表： kafka-topics.sh --list --bootstrap-server localhost:9092
   • 查看主题详情： kafka-topics.sh --describe --topic my-topic --bootstrap-server localhost:9092
   • 删除主题： kafka-topics.sh --delete --topic my-topic --bootstrap-server localhost:9092

2. kafka-console-producer.sh：用于从命令行向主题发送消息的命令。

   • 发送消息： kafka-console-producer.sh --topic my-topic --bootstrap-server localhost:9092

3. kafka-console-consumer.sh：用于从命令行消费主题中的消息的命令。

   • 消费消息： kafka-console-consumer.sh --topic my-topic --bootstrap-server localhost:9092

4. kafka-consumer-groups.sh：用于管理消费者组的命令。

   • 查看消费者组列表： kafka-consumer-groups.sh --list --bootstrap-server localhost:9092
   • 查看消费者组的消费情况： kafka-consumer-groups.sh --describe --group my-group --bootstrap-server localhost:9092

5. kafka-configs.sh：用于管理Kafka配置的命令。

   • 查看配置： kafka-configs.sh --describe --entity-type topics --entity-name my-topic --bootstrap-server localhost:9092
   • 修改配置： kafka-configs.sh --alter --entity-type topics --entity-name my-topic --add-config max.message.bytes=1048576 --bootstrap-server localhost:9092

这些是Kafka的一些常见命令，用于管理和操作Kafka集群、主题、消费者组等。你可以根据需要使用这些命令来管理你的Kafka环境。

🍁05. Kafka 和 Zookeeper 服务如何结合？

Kafka和Zookeeper是紧密结合的，它们在Kafka集群中扮演不同的角色。下面是Kafka和Zookeeper服务如何结合的概述：

1. Kafka依赖Zookeeper：Kafka通过Zookeeper来管理和维护集群的元数据信息、协调分区的领导者选举以及消费者组的协调工作。Kafka集群中的每个节点都会将自己的信息注册到Zookeeper中，并通过Zookeeper来协调和同步集群中的各个节点。

2. Zookeeper保存Kafka的元数据：Kafka的主题、分区、消费者组等元数据信息都存储在Zookeeper中。Kafka通过与Zookeeper的交互来获取和更新这些元数据信息，以保持集群的一致性和可靠性。

3. Leader选举：Kafka的每个分区都有一个分区领导者（Leader），负责处理该分区的读写请求。当分区的领导者不可用时，Zookeeper会协助进行新的领导者选举，选择一个新的分区领导者。

4. 消费者组协调：Kafka中的消费者可以组成消费者组，共同消费一个主题。Zookeeper负责协调消费者组的工作，包括分配分区给消费者、监控消费者的健康状态等。

总结来说，Kafka和Zookeeper通过密切的结合来实现Kafka集群的管理和协调。Kafka依赖于Zookeeper来管理集群的元数据、协调分区的领导者选举和消费者组的协调工作。因此，在使用Kafka时需要同时启动和管理Zookeeper服务，并确保Kafka集群和Zookeeper集群正常运行。

🍁06. Kafka 生产者和消费者的流程图？

🍁6.1 生产者流程图

看图识意：

🍁6.2 消费者流程图

看图识意：

🍁07. 安装 Zookeeper 服务的详细步骤？

安装Zookeeper服务的详细步骤如下：

1. 下载和解压Zookeeper：从Zookeeper官方网站（https://zookeeper.apache.org）下载适合你系统的Zookeeper版本.解压下载的文件到你选择的目录中.

2. 创建Zookeeper配置文件：在Zookeeper解压目录中，创建一个名为 zoo.cfg 的配置文件，并添加以下内容：

tickTime=2000dataDir=/path/to/zookeeper/dataclientPort=2181

将 /path/to/zookeeper/data 替换为你希望存储Zookeeper数据的路径。

1. 启动Zookeeper服务器：在Zookeeper解压目录中，执行以下命令启动Zookeeper服务器：

bin/zkServer.sh start

如果你想在后台运行Zookeeper服务器，可以使用以下命令：

bin/zkServer.sh start-foreground

1. 验证Zookeeper服务器：执行以下命令验证Zookeeper服务器是否成功启动：

bin/zkCli.sh

这将打开Zookeeper的命令行客户端。在客户端中，你可以执行一些Zookeeper命令来测试服务器的正常运行。

完成了以上步骤后，你就成功安装并启动了Zookeeper服务。你可以将Zookeeper与Kafka等其他系统结合使用，以实现分布式应用程序的管理和协调。记得根据你的需求进行配置和调整，以适应你的应用场景。

🍁08. 安装Kafka服务的详细步骤？

安装Kafka服务的详细步骤如下：

1. 安装Java：Kafka是用Java编写的，所以首先需要安装Java运行环境。确保你的系统上已经安装了Java，并且配置了正确的环境变量。

2. 下载和解压Kafka：从Kafka官方网站（https://kafka.apache.org/downloads）下载适合你系统的Kafka版本。解压下载的文件到你选择的目录中。

3. 配置Kafka：进入Kafka解压目录，编辑 config/server.properties 文件来配置Kafka。你可以根据需要进行配置，例如修改监听端口、日志存储路径等。

4. 启动Zookeeper：Kafka使用Zookeeper来管理集群的元数据信息。在Kafka解压目录中，执行以下命令启动Zookeeper服务器：

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

1. 启动Kafka服务器：在Kafka解压目录中，执行以下命令启动Kafka服务器：

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

1. 创建主题：使用 kafka-topics.sh 命令创建一个新的主题。例如，以下命令将创建一个名为 my-topic 的主题：

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

1. 测试Kafka：使用Kafka提供的命令行工具进行测试。可以使用 kafka-console-producer.sh 命令向主题发送消息，并使用 kafka-console-consumer.sh 命令从主题消费消息。

完成了以上步骤后，你就成功安装了Kafka服务，并可以开始使用它进行数据的生产和消费。记得根据你的具体需求进行配置和调整，以适应你的应用场景。

🍁09. Java项目结合Kafka服务使用。

🍁9.1 pom文件引入kafka服务

在Java项目中使用Kafka服务，你需要在项目的pom.xml文件中添加Kafka相关的依赖项。下面是一个示例pom.xml文件，展示如何引入Kafka服务的依赖项：

<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.example</groupId><artifactId>kafka-example</artifactId><version>1.0.0</version><dependencies><!-- Kafka dependencies --><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.8.0</version></dependency></dependencies></project>

在上述示例中，我们使用了Apache Kafka的kafka-clients依赖项。你可以根据需要选择适合的Kafka版本，并在 <dependencies> 标签中添加相应的依赖项。

完成pom.xml文件的修改后，保存并重新构建你的Java项目。这样，你的项目就可以使用Kafka相关的类和功能了。记得在代码中引入相应的Kafka包，以便使用Kafka的API进行生产和消费数据。

🍁9.2 创建Kafka消息发送组件

要在Java项目中创建Kafka消息发送组件，你可以使用Kafka提供的Java客户端API。以下是一个简单的示例，展示如何创建一个Kafka消息发送组件：

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.common.serialization.StringSerializer;import java.util.Properties;
import java.util.concurrent.Future;public class KafkaMessageSender {private KafkaProducer<String, String> producer;public KafkaMessageSender(String bootstrapServers) {Properties properties = new Properties();properties.put("bootstrap.servers", bootstrapServers);properties.put("key.serializer", StringSerializer.class.getName());properties.put("value.serializer", StringSerializer.class.getName());producer = new KafkaProducer<>(properties);}public void sendMessage(String topic, String message) {ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);Future<RecordMetadata> future = producer.send(record);// 可以根据需要处理发送结果}public void close() {producer.close();}
}

在上述示例中，我们创建了一个 KafkaMessageSender 类，它使用KafkaProducer来发送消息。在构造函数中，我们配置了Kafka的连接信息和序列化器。 sendMessage 方法用于发送消息到指定的主题，你可以根据需要进行处理发送结果。 close 方法用于关闭KafkaProducer。

要使用这个Kafka消息发送组件，你可以在你的Java项目中创建一个实例，并调用 sendMessage 方法发送消息到Kafka主题。记得在创建组件实例时，提供正确的Kafka服务器地址（bootstrap.servers）。

public class Main {public static void main(String[] args) {KafkaMessageSender sender = new KafkaMessageSender("localhost:9092");sender.sendMessage("my-topic", "Hello Kafka!");sender.close();}
}

在上述示例中，我们创建了一个 Main 类，创建了一个 KafkaMessageSender 实例，并调用 sendMessage 方法发送消息到名为 my-topic 的Kafka主题。

🍁9.3 创建Kafka消息消费组件

要在Java项目中创建Kafka消息消费组件，你可以使用Kafka提供的Java客户端API。以下是一个简单的示例，展示如何创建一个Kafka消息消费组件：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class KafkaMessageConsumer {private KafkaConsumer<String, String> consumer;public KafkaMessageConsumer(String bootstrapServers, String groupId) {Properties properties = new Properties();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());consumer = new KafkaConsumer<>(properties);}public void consumeMessages(String topic) {consumer.subscribe(Collections.singletonList(topic));while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {System.out.println("Received message: " + record.value());// 可以根据需要处理接收到的消息}}}public void close() {consumer.close();}
}

在上述示例中，我们创建了一个 KafkaMessageConsumer 类，它使用KafkaConsumer来消费消息。在构造函数中，我们配置了Kafka的连接信息、消费者组ID以及序列化器。 consumeMessages 方法用于订阅指定的主题并消费消息。在消费循环中，我们使用 poll 方法来拉取消息，并遍历处理接收到的消息。你可以根据需要对消息进行处理。 close 方法用于关闭KafkaConsumer。

要使用这个Kafka消息消费组件，你可以在你的Java项目中创建一个实例，并调用 consumeMessages 方法来消费指定主题的消息。记得在创建组件实例时，提供正确的Kafka服务器地址（bootstrap.servers）和消费者组ID（group.id）。

public class Main {public static void main(String[] args) {KafkaMessageConsumer consumer = new KafkaMessageConsumer("localhost:9092", "my-group");consumer.consumeMessages("my-topic");// 在消费完成后记得调用 close 方法关闭消费者consumer.close();}
}

在上述示例中，我们创建了一个 Main 类，创建了一个 KafkaMessageConsumer 实例，并调用 consumeMessages 方法来消费名为 my-topic 的Kafka主题的消息。

🍁9.4 Java项目使用Kafka小结

使用Kafka的Java项目需要以下步骤：

1. 下载和安装Kafka：从Kafka官方网站下载适合你系统的Kafka版本，并按照官方文档进行安装。

2. 配置Kafka：编辑Kafka的配置文件，指定Kafka的监听地址、日志存储路径等配置项。

3. 创建生产者：使用Kafka提供的Java客户端API，创建一个Kafka生产者，配置生产者的连接信息和序列化器。

4. 发送消息：使用生产者发送消息到指定的Kafka主题。

5. 创建消费者：使用Kafka提供的Java客户端API，创建一个Kafka消费者，配置消费者的连接信息和反序列化器。

6. 订阅主题：让消费者订阅感兴趣的Kafka主题。

7. 消费消息：从订阅的主题中轮询消费消息，并处理接收到的消息。

8. 关闭生产者和消费者：在程序结束时，关闭生产者和消费者，释放资源。

使用Kafka的Java项目可以实现高吞吐量、可靠性和实时处理的数据流处理。Kafka提供了丰富的API和工具，使得在Java项目中使用Kafka变得简单和灵活。通过将消息发送到Kafka主题并从主题中消费消息，你可以构建实时流处理应用程序、日志收集系统、事件驱动架构等。

需要注意的是，Kafka的配置和管理可能需要一定的学习和经验。在使用Kafka时，建议参考官方文档和示例代码，以确保正确配置和使用Kafka的各项功能。