kafka小白基础知识

一、Kafka 入门

（一）Kafka 简介

Kafka 是一个开源的分布式流处理平台，最初由 LinkedIn 开发，后来贡献给了 Apache 软件基金会。它被设计用于处理实时数据流，具有高吞吐量、可扩展性、持久性和容错性等特点。Kafka 主要用于构建实时数据管道和流式应用程序，如日志收集、消息系统、事件驱动架构等。

（二）核心概念

主题（Topic）：Kafka 中的消息以主题为单位进行分类，类似于数据库中的表。生产者将消息发送到特定的主题，消费者从主题中读取消息。
分区（Partition）：每个主题可以被划分为多个分区，分区是 Kafka 实现高吞吐量和可扩展性的关键。消息在分区内是有序的，但在整个主题中不一定有序。
生产者（Producer）：负责将消息发送到 Kafka 主题的应用程序。生产者可以根据需要选择将消息发送到特定的分区。
消费者（Consumer）：从 Kafka 主题中读取消息的应用程序。消费者可以以组的形式存在，同一个组内的消费者共同消费主题中的消息，每个分区只能被组内的一个消费者消费。
代理（Broker）：Kafka 集群中的每个服务器节点称为代理。代理负责存储和管理分区的数据，并处理生产者和消费者的请求。

（三）安装与启动

1. 下载 Kafka

从 Apache Kafka 官方网站（Apache Kafka）下载最新版本的 Kafka。

2. 解压文件

bash

tar -zxvf kafka_2.13 - 3.4.0.tgz
cd kafka_2.13 - 3.4.0

3. 单机安装与启动

启动 ZooKeeper

Kafka 使用 ZooKeeper 来管理集群的元数据，首先启动 ZooKeeper：

bash

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

启动 Kafka 代理

bash

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

4. 集群安装与启动

环境准备

假设我们要搭建一个包含 3 个节点的 Kafka 集群，节点的 IP 地址分别为 192.168.111.2、192.168.111.3、192.168.111.4。在每个节点上都需要完成 Kafka 的下载和解压操作。

修改配置文件

在每个节点上修改 config/server.properties 文件：

节点 1（192.168.111.2）

properties

broker.id=0
listeners=PLAINTEXT://192.168.111.2:9092
advertised.listeners=PLAINTEXT://192.168.111.2:9092
zookeeper.connect=192.168.111.2:2181,192.168.111.3:2181,192.168.111.4:2181

节点 2（192.168.111.3）

properties

broker.id=1
listeners=PLAINTEXT://192.168.111.3:9092
advertised.listeners=PLAINTEXT://192.168.111.3:9092
zookeeper.connect=192.168.111.2:2181,192.168.111.3:2181,192.168.111.4:2181

节点 3（192.168.111.4）

properties

broker.id=2
listeners=PLAINTEXT://192.168.111.4:9092
advertised.listeners=PLAINTEXT://192.168.111.4:9092
zookeeper.connect=192.168.111.2:2181,192.168.111.3:2181,192.168.111.4:2181

配置说明：

broker.id：每个 Kafka 代理的唯一标识符，不能重复。
listeners：代理监听的地址和端口。
advertised.listeners：对外公布的地址和端口，用于生产者和消费者连接。
zookeeper.connect：ZooKeeper 集群的连接地址。

启动 ZooKeeper 集群

在每个节点上启动 ZooKeeper：

bash

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

启动 Kafka 集群

在每个节点上启动 Kafka 代理：

bash

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

（四）创建主题

使用 Kafka 提供的命令行工具创建一个主题：

bash

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

--topic：指定主题名称
--bootstrap-server：指定 Kafka 代理的地址
--partitions：指定主题的分区数
--replication-factor：指定分区的副本数

（五）发送和接收消息

1. 启动生产者

bash

bin/kafka-console-producer.sh --topic test_topic --bootstrap-server 192.168.111.2:9092

在控制台输入消息，按回车键发送。

2. 启动消费者

bash

bin/kafka-console-consumer.sh --topic test_topic --from-beginning --bootstrap-server 192.168.111.2:9092

--from-beginning 表示从主题的开头开始消费消息。

二、Kafka 的安全性

（一）身份验证

Kafka 支持多种身份验证机制，如 SSL/TLS、SASL（Simple Authentication and Security Layer）等。

1. SSL/TLS 身份验证

配置步骤：
- 生成证书和密钥，包括 CA 证书、服务器证书和客户端证书。
- 在 Kafka 代理的 server.properties 中配置 SSL 相关参数，如 listeners、ssl.keystore.location、ssl.keystore.password 等。
- 在生产者和消费者的配置中也需要配置相应的 SSL 参数，如 security.protocol=SSL、ssl.truststore.location、ssl.truststore.password 等。

2. SASL 身份验证

SASL 提供了多种认证机制，如 PLAIN、GSSAPI（用于 Kerberos 认证）等。以 PLAIN 机制为例：

配置步骤：
- 在 Kafka 代理的 server.properties 中配置 SASL 相关参数，如 listeners=SASL_PLAINTEXT://:9092、sasl.enabled.mechanisms=PLAIN 等。
- 创建用户和密码文件，并在 server.properties 中指定文件路径。
- 生产者和消费者配置相应的 SASL 参数，如 security.protocol=SASL_PLAINTEXT、sasl.mechanism=PLAIN 等。

（二）授权

Kafka 可以通过 ACL（Access Control List）来控制用户对主题、分区等资源的访问权限。

配置步骤

启用 ACL：在 Kafka 代理的 server.properties 中设置 authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer。
创建 ACL 规则：使用 Kafka 提供的命令行工具 kafka-acls.sh 来创建和管理 ACL 规则。例如，允许用户 user1 对主题 test_topic 进行读写操作：

bash

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.111.2:2181 --add --allow-principal User:user1 --operation Read --operation Write --topic test_topic

（三）数据加密

Kafka 可以使用 SSL/TLS 对数据进行加密传输，确保数据在网络传输过程中的安全性。配置 SSL/TLS 加密的步骤与 SSL/TLS 身份验证类似，主要是在代理、生产者和消费者的配置中设置相关的 SSL 参数。

三、Kafka 的特性

（一）高吞吐量

Kafka 的设计目标之一是实现高吞吐量的数据处理。它采用了批量处理、顺序读写磁盘、零拷贝等技术，能够在短时间内处理大量的消息。例如，在一个高并发的日志收集场景中，Kafka 可以轻松应对每秒数万条甚至更多的日志消息。

（二）可扩展性

Kafka 可以通过增加代理节点来扩展集群的规模，以应对不断增长的数据量和并发请求。新的代理节点可以无缝加入集群，并且 Kafka 会自动进行分区的重新分配，确保数据的均匀分布。

（三）持久性和容错性

Kafka 将消息持久化存储在磁盘上，并且支持消息的多副本复制。每个分区可以有多个副本，分布在不同的代理节点上。当某个代理节点出现故障时，其他副本可以继续提供服务，保证数据的可用性和持久性。

（四）分布式特性

Kafka 是一个分布式系统，各个代理节点之间通过 ZooKeeper 进行协调和管理。生产者和消费者可以分布式部署，并行地进行消息的生产和消费，提高系统的整体性能和可靠性。

四、保障 Kafka 数据一致性

（一）副本机制

Kafka 通过副本机制来保障数据的一致性和可用性。每个分区可以有多个副本，其中一个副本作为领导者（Leader），负责处理所有的读写请求；其他副本作为追随者（Follower），从领导者副本同步数据。

工作原理

生产者将消息发送到领导者副本，领导者副本将消息写入本地日志，并向追随者副本进行同步。
追随者副本从领导者副本拉取消息，并写入本地日志。当追随者副本确认收到消息后，会向领导者副本发送确认信息。
领导者副本只有在收到大多数副本（超过半数）的确认信息后，才会向生产者返回消息写入成功的响应。

（二）ISR（In-Sync Replicas）机制

ISR 是指与领导者副本保持同步的追随者副本集合。Kafka 通过 ISR 机制来确保数据的一致性和可用性。

工作原理

领导者副本会定期检查追随者副本的同步状态，如果某个追随者副本的同步延迟超过一定阈值，领导者副本会将其从 ISR 中移除。
只有当消息被写入 ISR 中的所有副本后，才被认为是已提交的消息。消费者只能消费已提交的消息，从而保证了数据的一致性。

（三）acks 参数

生产者在发送消息时，可以通过设置 acks 参数来控制消息的确认机制，从而影响数据的一致性。

参数取值及含义

acks=0：生产者发送消息后，不需要等待任何确认信息，直接认为消息发送成功。这种方式吞吐量最高，但可能会丢失消息，数据一致性最差。
acks=1：生产者发送消息后，只需要等待领导者副本确认收到消息，就认为消息发送成功。这种方式在一定程度上保证了数据的一致性，但如果领导者副本在消息同步给追随者副本之前出现故障，可能会丢失消息。
acks=all 或 acks=-1：生产者发送消息后，需要等待 ISR 中的所有副本都确认收到消息，才认为消息发送成功。这种方式数据一致性最高，但吞吐量相对较低。