Kafka快速入门

Posted 2023-03-08 一颗奋起萌发的种子

文章目录

• 一、Kafka 概述
• • 1.1 主题（Topic）
  • 1.2 生产者（Producer）
  • 1.3 消费者（Consumer）
  • 1.4 分区（Partition）
  • 1.5 消费者组（Consumer Group）
  • 1.6 代理（Broker）
  • 1.7 ZooKeeper
• 二、Kafka 的安装和配置
• 三、Kafka 的基本操作
• • 3.1 创建主题
  • 3.2 发布消息
  • 3.3 消费消息
• 四、总结

一、Kafka 概述

Kafka 是一种高吞吐量、分布式的消息队列系统，由 LinkedIn 公司开发。它最初是为了解决 LinkedIn 在海量实时数据处理场景下的消息传递问题而设计的。Kafka 提供了一种可靠的、分布式的、可分区的、可重复消费的消息发布与订阅系统。它允许生产者将消息发布到一个或多个主题（topic）中，并允许消费者订阅一个或多个主题并消费发布到这些主题的消息。Kafka 的消息处理方式非常高效，每秒可以处理数百万个消息。

1.1 主题（Topic）

Kafka 消息被发布到主题（Topic）中，主题可以被认为是一类消息的集合。每个主题都有一个唯一的名称，生产者将消息发布到主题中，而消费者则订阅感兴趣的主题并消费其中的消息。

1.2 生产者（Producer）

生产者（Producer）是将消息发布到 Kafka 主题中的客户端应用程序。生产者可以将消息发布到一个或多个主题中，并且可以根据需要指定消息的键（Key），Kafka 会根据键值将消息分配到相应的分区（Partition）中。

1.3 消费者（Consumer）

消费者（Consumer）是从 Kafka 主题中订阅并消费消息的客户端应用程序。消费者可以订阅一个或多个主题，并从中消费消息。消费者可以控制消息的读取进度，并可以按照需要重复消费消息。

1.4 分区（Partition）

每个主题（Topic）都可以被分为一个或多个分区（Partition），每个分区都有一个唯一的标识符（Partition ID）。Kafka 在分区上进行消息存储和处理，并可以根据生产者指定的键（Key）进行有序分发。每个分区在 Kafka 集群中都有多个副本，以提高可靠性和容错性。

1.5 消费者组（Consumer Group）

消费者组（Consumer Group）是一组消费者的集合，它们共同订阅一个或多个主题（Topic）。Kafka 将主题中的消息均匀分配给每个消费者组中的消费者，以实现负载均衡和高可用性。

1.6 代理（Broker）

Kafka 集群中的每个服务器节点都称为 Broker，它们负责存储和处理分区中的消息。每个分区在 Kafka 集群中都有多个 Broker 副本，它们之间通过 ZooKeeper 协调，以确保消息的可靠性和容错性。

1.7 ZooKeeper

ZooKeeper 是一个开源的分布式协调系统，Kafka 使用 ZooKeeper 来管理 Broker 和消费者的状态，并协调它们之间的通信。ZooKeeper 在 Kafka 中扮演了重要的角色，它负责管理 Kafka 集群的元数据信息、维护分区的状态信息和负载均衡、管理消费者组的状态等。

二、Kafka 的安装和配置

在开始使用 Kafka 之前，你需要先下载并安装 Kafka，并对其进行必要的配置。你可以按照以下步骤完成 Kafka 的安装和配置：

1. 下载 Kafka：你可以在 Kafka 的官方网站上下载最新的 Kafka 版本。

2. 解压 Kafka：将下载的 Kafka 压缩包解压到你希望安装 Kafka 的目录下。

   $ tar -xzf kafka_2.13-3.4.0.tgz
   $ cd kafka_2.13-3.4.0
   

3. 配置 Kafka：Kafka 的配置文件位于解压后的目录中的 config 目录下。你需要修改 server.properties 文件，配置 Kafka 的基本参数，比如监听地址、端口号、分区数、副本数等。

4. 启动 Kafka：Kafka 启动脚本位于解压后的目录中的 bin 目录下，你可以运行以下命令启动 Kafka：
   注意：您的本地环境必须安装 Java 8+。
   先启动zookeeper服务：

   # Start the ZooKeeper service
   $ bin/zookeeper-server-start.sh config/zookeeper.properties
   

   如果安装的Kafka版本过旧，没有zookeeper，也可以自行下载并安装好zookeeper。

   再启动Kafka服务：

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

   启动 Kafka 后，你可以使用以下命令验证 Kafka 是否正常工作：

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

   如果能够输出 Kafka 中所有的主题列表，则表示 Kafka 正常工作。

三、Kafka 的基本操作

Kafka 提供了丰富的命令行工具，可以帮助你管理 Kafka 的主题、生产者、消费者、分区等资源。下面我们将介绍 Kafka 的一些常用命令行操作。

3.1 创建主题

要创建一个主题，你可以使用 kafka-topics.sh 工具。以下命令可以创建一个名为 test-topic 的主题：

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

其中，--topic 参数指定了要创建的主题名称，--partitions 参数指定了主题的分区数，--replication-factor 参数指定了每个分区的副本数。在创建主题时，你还需要指定 Kafka 的地址，可以使用 --bootstrap-server 参数指定。

3.2 发布消息

要向主题中发布消息，你需要使用生产者客户端程序。以下是一个简单的 Java 生产者示例：

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class ProducerDemo 
    public static void main(String[] args) 
        String topicName = "test-topic";
        String bootstrapServer = "localhost:9092";

        // 设置生产者客户端的配置信息
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServer);
        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);

        // 构造消息
        String messageKey = "key1";
        String messageValue = "hello world";

        // 发送消息
        ProducerRecord<String, String> record = new ProducerRecord<>(topicName, messageKey, messageValue);
        producer.send(record);

        // 关闭生产者客户端
        producer.close();
    

以上代码创建了一个生产者客户端，并向名为 test-topic 的主题中发送了一条消息。

3.3 消费消息

要从主题中消费消息，你需要使用消费者客户端程序。以下是一个简单的 Java 消费者示例：

import org.apache.kafka.clients.consumer.*;
import java.util.Collections;
import java.util.Properties;

public class ConsumerDemo 
    public static void main(String[] args) 
        String topicName = "test-topic";
        String bootstrapServer = "localhost:9092";
        String groupId = "test-group";

        // 设置消费者客户端的配置信息
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServer);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("group.id", groupId);

        // 创建消费者客户端
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList(topicName));

        // 消费消息
        while (true) 
            ConsumerRecords<String, String> records = consumer.poll(1000);
            for (ConsumerRecord<String, String> record : records) 
                System.out.printf("topic=%s, partition=%d, offset=%d, key=%s, value=%s%n",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value());
            
        
    

以上代码创建了一个消费者客户端，并从名为 test-topic 的主题中消费了所有消息。你可以使用以下命令行工具来验证消费者是否正常工作：

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

以上命令将从 test-topic 主题中消费所有消息。

四、总结

本文简要介绍了 Kafka 的基本概念和使用方法。Kafka 是一个高吞吐量、分布式的消息传递系统，它可以用于构建实时数据流处理应用程序。Kafka 提供了丰富的命令行工具，可以帮助你管理 Kafka 的主题、生产者、消费者等。同时，Kafka 也提供了 Java、Python、Scala、C++ 等多种编程语言的客户端库，方便开发者在自己熟悉的编程语言中使用 Kafka。

希望本文对你了解 Kafka 有所帮助。如果你对 Kafka 有更深入的了解，可以阅读官方文档或相关书籍，进一步深入学习。