Kafka资料入门教程：轻松上手Apache Kafka

本文全面介绍了Kafka，包括其定义、主要特点、应用场景以及详细的架构详解。文章还提供了Kafka的安装指南，涵盖从准备环境到实际部署的全过程，并深入解析了Kafka的核心概念，如Topic与Partition、Producer与Consumer，并提供了操作实践和常见问题解答。

Apache Kafka 是一种高吞吐量的分布式流处理平台，最初由 LinkedIn 开发，后成为 Apache 软件基金会的顶级项目。Kafka 作为流处理系统的底层，可以用于构建实时数据管道和流式应用程序。它提供了一个高并发、低延迟的消息传递系统，可以轻松地将数据从一个系统转移到另一个系统，同时支持数千个消费者和生产者。

• 高吞吐量：Kafka 设计用于处理每秒数千到数百万的消息，使其成为需要高性能数据传输的应用的理想选择。
• 持续可用性：通过复制消息到多个节点，Kafka 实现了高可用性，即使在出现故障时也能确保消息不丢失。
• 持久性：消息在 Kafka 中被持久化存储，支持长期数据存储需求。
• 水平扩展：Kafka 可轻松扩展，通过添加更多机器来处理更多负载。
• 低延迟：即使在高吞吐量情况下，Kafka 也能保证非常低的消息传递延迟。
• 支持多种数据格式：Kafka 支持多种数据格式，包括结构化和非结构化的数据。

• 日志聚合：将多个应用的日志合并为一个日志流，便于集中分析和查询。
• 指标聚合：收集来自不同来源的指标数据，例如监控工具、应用服务器等，以便进行实时分析。
• 流处理：将数据流进行转换、过滤或聚合，生成新的数据流。
• 数据管道：从一个系统或服务实时地将数据传输到另一个系统或服务。
• 异步消息传递：应用之间进行异步通信，解耦业务系统。
• 数据集成：将多个数据源的数据整合到一个统一的数据平台中，实现数据的统一管理。

Kafka 集群由多个节点组成，每个节点都是一个 Broker，它们之间通过网络通信。每个 Broker 都可以处理多个 Topic（主题），而每个 Topic 又可以包含多个 Partition（分区）。Partition 使得数据可以在不同的 Broker 之间进行负载均衡，从而实现水平扩展。

• Broker：Kafka 集群中的每个节点都称为一个 Broker，它们负责存储消息和处理来自 Producer 和 Consumer 的请求。
• Topic：Topic 是消息的逻辑分类，可以理解为一个消息队列。每个 Topic 由多个 Partition 组成。
• Partition：Partition 是 Topic 的逻辑分片，每一个 Partition 都是有序且不可变的消息序列。
• Producer：Producer 负责将消息发送到 Kafka Broker，可以将消息发送到指定的 Topic。
• Consumer：Consumer 负责从 Kafka Broker 中读取消息。Consumer 可以加入一个 Consumer Group 来处理同一个 Topic 的多个 Partition。
• ZooKeeper：Kafka 使用 ZooKeeper 来管理集群的元数据和配置信息，包括 Topic 的配置、Partition 的分配以及 Consumer Group 的状态等。

Kafka 可以与多种技术集成，包括：

• Spark Streaming：使用 Spark Streaming 处理 Kafka 中的消息。
• Storm：使用 Storm 处理实时流处理。
• Hadoop：将 Kafka 数据导入 Hadoop 进行批处理分析。
• Elasticsearch：将 Kafka 数据实时导入 Elasticsearch 用于搜索和分析。
• HBase：将 Kafka 数据实时导入 HBase 用于实时数据存储。
• Flume：使用 Flume 作为 Kafka 和 Hadoop 之间的桥梁。
• Kinesis：AWS 的流数据处理服务，可以与 Kafka 无缝集成。
• Apache Flink：使用 Flink 处理 Kafka 中的实时数据流。
• Logstash：使用 Logstash 将 Kafka 中的消息发送到 Elasticsearch 或其他数据存储。

在安装 Kafka 之前，需要确保系统已经安装了 Java 和 ZooKeeper。Kafka 依赖于 Java，因此安装 Java 是必需的。Kafka 使用 ZooKeeper 来管理集群的元数据和配置信息，因此也需要安装 ZooKeeper。

安装Java

可以在官方网站上下载 Java 的安装包，这里我们使用 Java 8 作为示例。

# 下载 Java 8
wget --no-check-certificate --no-cookies --header "Cookie: oraclelicense=accept-securebackup-cookie" http://download.oracle.com/otn-pub/java/jdk/8u221-b11/1c4c45a38bbe448f855aa627e90a0532/jdk-8u221-linux-x64.tar.gz

# 解压 Java 安装包
tar -zxvf jdk-8u221-linux-x64.tar.gz

# 将 Java 安装到 /usr/local
sudo mv jdk1.8.0_221 /usr/local/java

# 设置环境变量
export JAVA_HOME=/usr/local/java/jdk1.8.0_221
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar

安装ZooKeeper

可以在 Apache ZooKeeper 的官方网站上下载 ZooKeeper 的安装包，这里我们使用 ZooKeeper 3.5.6 作为示例。

# 下载 ZooKeeper 3.5.6
wget http://mirror.beyondhosting.net/apache/zookeeper/stable/zookeeper-3.5.6.tar.gz

# 解压 ZooKeeper 安装包
tar -zxvf zookeeper-3.5.6.tar.gz

# 将 ZooKeeper 安装到 /usr/local
sudo mv zookeeper-3.5.6 /usr/local/zookeeper

# 设置环境变量
export ZOOKEEPER_HOME=/usr/local/zookeeper/zookeeper-3.5.6
export PATH=$ZOOKEEPER_HOME/bin:$PATH
export CLASSPATH=.:$ZOOKEEPER_HOME/bin:$ZOOKEEPER_HOME/conf:$ZOOKEEPER_HOME/zookeeper-3.5.6.jar

下载Kafka

可以在 Apache Kafka 的官方网站上下载 Kafka 的安装包，这里我们使用 Kafka 2.8.0 作为示例。

# 下载 Kafka 2.8.0
wget http://mirror.bjtu.edu.cn/apache/kafka/2.8.0/kafka_2.13-2.8.0.tgz

# 解压 Kafka 安装包
tar -zxvf kafka_2.13-2.8.0.tgz

# 将 Kafka 安装到 /usr/local
sudo mv kafka_2.13-2.8.0 /usr/local/kafka

配置Kafka

配置 Kafka 的安装目录下的 server.properties 文件。需要修改的配置项包括 broker.id（每个 Kafka Broker 的唯一标识）、listeners（监听地址）、log.dirs（日志存储路径）以及 zookeeper.connect（连接 ZooKeeper 的地址）。

# 配置文件 server.properties
broker.id=0
listeners=PLAINTEXT://localhost:9092
log.dirs=/usr/local/kafka/kafka_2.13-2.8.0/kafka-logs
zookeeper.connect=localhost:2181

启动ZooKeeper

在启动 Kafka 之前，需要启动 ZooKeeper。

# 进入 ZooKeeper 安装目录
cd /usr/local/zookeeper/zookeeper-3.5.6

# 启动 ZooKeeper
bin/zkServer.sh start

启动Kafka Broker

在启动 Kafka Broker 时，需要指定配置文件的路径。

# 进入 Kafka 安装目录
cd /usr/local/kafka/kafka_2.13-2.8.0

# 启动 Kafka Broker
bin/kafka-server-start.sh -daemon /usr/local/kafka/kafka_2.13-2.8.0/config/server.properties

下载Kafka

可以在 Apache Kafka 的官方网站上下载 Kafka 的安装包，这里我们使用 Kafka 2.8.0 作为示例。下载后解压到 C:\kafka。

配置Kafka

配置 Kafka 的安装目录下的 server.properties 文件。需要修改的配置项包括 broker.id（每个 Kafka Broker 的唯一标识）、listeners（监听地址）、log.dirs（日志存储路径）以及 zookeeper.connect（连接 ZooKeeper 的地址）。

# 配置文件 server.properties
broker.id=0
listeners=PLAINTEXT://localhost:9092
log.dirs=C:\kafka\kafka_2.13-2.8.0\kafka-logs
zookeeper.connect=localhost:2181

启动ZooKeeper

在启动 Kafka 之前，需要启动 ZooKeeper。ZooKeeper 的下载和配置与 Linux 环境类似，但在 Windows 中需要将其解压到 C:\zookeeper 并进行配置。

# 进入 ZooKeeper 安装目录
cd C:\zookeeper\zookeeper-3.5.6

# 启动 ZooKeeper
.\bin\zkServer.cmd

启动Kafka Broker

在启动 Kafka Broker 时，需要指定配置文件的路径。

# 进入 Kafka 安装目录
cd C:\kafka\kafka_2.13-2.8.0

# 启动 Kafka Broker
.\bin\windows\kafka-server-start.bat config\server.properties

Topic

Topic 是 Kafka 中消息的逻辑分类，可以理解为一个消息队列。每个 Topic 都可以包含多个 Partition，Partition 是 Topic 的逻辑分片，每一个 Partition 都是有序且不可变的消息序列。每个 Partition 都是一个独立的数据结构，可以存储大量的消息，并且可以在不同的 Broker 上进行负载均衡。

Partition

每个 Partition 都是由一系列有序且不可变的消息组成的日志文件。每个 Partition 都会有一个 Leader 和多个 Follower。Leader 负责处理来自 Producer 和 Consumer 的请求，而 Follower 则负责复制 Leader 的数据。每个 Partition 的消息都会被分配一个唯一的偏移量（Offset），用于标识每条消息的位置。

Producer

Producer 负责将消息发送到 Kafka Broker。Producer 可以将消息发送到指定的 Topic，每个 Producer 都会将消息发送到 Topic 的某个 Partition 中。为了保证消息的顺序性，可以配置 Producer 发送消息到特定的 Partition，也可以让 Kafka 自动分配。

Consumer

Consumer 负责从 Kafka Broker 中读取消息。Consumer 可以加入一个 Consumer Group 来处理同一个 Topic 的多个 Partition。每个 Consumer Group 都有一个唯一的标识符，用于区分不同的消费组。Consumer 会从 Topic 中的 Partition 中读取消息，每个 Consumer 会从一个或多个 Partition 中读取消息。Consumer 可以通过配置来控制它读取消息的速率。

Consumer Group 是一个逻辑的概念，用于区分不同的消费组。每个 Consumer 都属于一个 Consumer Group，不同的 Consumer Group 可以消费同一个 Topic，但是每个 Consumer Group 只能消费 Topic 中的一个 Partition。当一个 Consumer Group 中的 Consumer 数量增加时，Kafka 会自动将 Topic 中的 Partition 分配给新的 Consumer。当一个 Consumer Group 中的 Consumer 数量减少时，Kafka 会自动将 Topic 中的 Partition 从死的 Consumer 上移除。

创建Topic

首先，需要创建一个 Topic。可以使用 kafka-topics.sh 脚本来创建 Topic。

# 创建一个名为 test 的 Topic
bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1

发送消息

接下来，可以使用 kafka-console-producer.sh 脚本来发送消息。

# 发送消息到 test Topic
bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:9092

在命令行中输入消息，每输入一条消息就按回车键发送，例如：

hello kafka
world

创建Consumer Group

首先，需要创建一个 Consumer Group。可以使用 kafka-console-consumer.sh 脚本来创建 Consumer Group。

# 创建一个名为 group1 的 Consumer Group
bin/kafka-console-consumer.sh --topic test --from-beginning --bootstrap-server localhost:9092 --group group1

消费消息

接下来，可以使用 kafka-console-consumer.sh 脚本来消费消息。

# 消费 test Topic 中的消息
bin/kafka-console-consumer.sh --topic test --bootstrap-server localhost:9092 --from-beginning --group group1

在命令行中可以看到之前发送的消息，例如：

hello kafka
world

持久化消息

Kafka 会将消息持久化到磁盘上，可以通过配置 log.retention.hours 来设置消息的保留时间。默认情况下，消息会被保留 7 天。可以通过以下命令来查询 Topic 的配置。

# 查询 test Topic 的配置
bin/kafka-configs.sh --describe --topic test --bootstrap-server localhost:9092

查询消息

可以通过 kafka-run-class.sh 脚本来查询消息。

# 查询 test Topic 中的消息
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic test --time -1 --max-messages 1 --bootstrap-server localhost:9092

在命令行中可以看到消息的偏移量和时间戳，例如：

test:0:1633428480000

错误代码：KAFKA-1

错误代码 KAFKA-1 表示 Broker 出现了错误，可能是由于配置错误或网络连接问题。检查 Broker 的配置文件，确保 listeners 和 zookeeper.connect 配置正确。检查是否已经启动了 ZooKeeper 服务，并确保 Broker 能够连接到 ZooKeeper。

错误代码：KAFKA-2

错误代码 KAFKA-2 表示 Producer 发送消息时出现了错误，可能是由于网络连接问题或 Broker 未启动。检查网络连接，确保 Broker 已经启动并监听了正确的端口。

错误代码：KAFKA-3

错误代码 KAFKA-3 表示 Consumer 读取消息时出现了错误，可能是由于网络连接问题或 Broker 未启动。检查网络连接，确保 Broker 已经启动并监听了正确的端口。

错误代码：KAFKA-4

错误代码 KAFKA-4 表示 Consumer Group 出现了错误，可能是由于配置错误或网络连接问题。检查 Consumer Group 的配置，确保 group.id 配置正确。检查网络连接，确保 Broker 已经启动并监听了正确的端口。

调整消息压缩

Kafka 支持消息压缩，可以通过设置 compression.type 配置项来启用消息压缩。启用消息压缩可以减少网络传输的带宽，提高消息传输的效率。

# 启用消息压缩
compression.type=producer

调整消息批处理

Kafka 支持消息批处理，可以通过设置 batch.size 和 linger.ms 配置项来调整消息批处理的大小和延迟。调整消息批处理的大小和延迟可以减少网络传输的次数，提高消息传输的效率。

# 调整消息批处理的大小和延迟
batch.size=16384
linger.ms=5

调整消息缓存

Kafka 支持消息缓存，可以通过设置 buffer.memory 配置项来调整消息缓存的大小。调整消息缓存的大小可以提高消息传输的效率。

# 调整消息缓存的大小
buffer.memory=33554432

调整消息持久化

Kafka 支持消息持久化，可以通过设置 log.retention.hours 和 log.segment.bytes 配置项来调整消息持久化的大小和时间。调整消息持久化的大小和时间可以减少磁盘空间的使用，提高消息持久化的效率。

# 调整消息持久化的大小和时间
log.retention.hours=168
log.segment.bytes=104857600

日志管理

Kafka 支持日志管理，可以通过设置 log4j.properties 文件来配置日志的级别和输出位置。通过日志管理可以更好地监控 Kafka 的运行状态。

# 配置日志的级别和输出位置
log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %c{1}:%L - %m%n

监控

Kafka 支持监控，可以通过设置 metricsContext 配置项来启用监控。通过监控可以更好地了解 Kafka 的运行状态。

# 启用监控
metricsContext=JMX

可以通过 JMX 来监控 Kafka 的运行状态，例如：

# 监控 Kafka 的运行状态
jmxquery -m kafka.consumer:type=*,kafka.producer:type=*,kafka.server:type=*

可以通过 Kafka 自带的监控工具 kafka-run-class.sh 来监控 Kafka 的运行状态，例如：

# 监控 Kafka 的运行状态
bin/kafka-run-class.sh kafka.tools.JmxTool --jmx-service-url service:jmx:rmi:///jndi/rmi://localhost:9999/jmxrmi --object-name kafka.consumer:type=*,kafka.producer:type=*,kafka.server:type=*

可以通过第三方监控工具来监控 Kafka 的运行状态，例如：

# 监控 Kafka 的运行状态
./grafana-cli plugins install grafana-kafka-datasource
./grafana-cli plugins install grafana-kafka-panel