Kafka系列教程 - Kafka 快速入门 -1

1. Kafka 简介

官网定义：

Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications.

Apache Kafka是一个开源的分布式事件流式平台，被数千个公司用于高性能的数量管道，流式数据分析，数据集成和关键任务应用程序。

Kafka由LinkedIn公司开发并于2011年早期开源，2012年10月23日从Apache Incubator毕业。

1.1. Kafka 的功能

Kafka 的核心功能如下：

• 消息引擎 - Kafka 可以作为一个消息引擎系统。
• 流处理 - Kafka 可以作为一个分布式流处理平台。
• 存储 - Kafka 可以作为一个安全的分布式存储。

另外还有：

1. 高吞吐量和低延迟

• Kafka 设计支持高吞吐量，即使在处理数百万条消息时，性能依然稳定。
• 其日志结构和零拷贝机制优化了 I/O 操作，实现了低延迟的数据传输。

---

2. 持久化消息存储

• Kafka 使用分布式日志存储机制，所有消息都写入磁盘并可配置存储时间。
• 它提供持久化和可靠性保障，即使在消费者处理失败时，数据依然可用。

---

3. 分布式架构

• Kafka 运行在一个分布式集群上，具有高扩展性和高可用性。
• Topic 被分片为多个分区（Partition），分布在不同的 Broker 上，允许并行处理和存储。

---

4. 发布/订阅模型

• Kafka 支持发布/订阅（Pub/Sub）模式，允许生产者向特定的 Topic 发布消息，消费者可以订阅并处理这些消息。
• 消费者组（Consumer Groups）机制确保了消息的灵活分发和负载均衡。

---

5. 消息顺序性

• Kafka 保证单个分区内消息的顺序性，使得在某些场景下可以精确重现事件流。

---

6. 高容错性

• Kafka 支持多副本机制（Replication），每个分区的数据会复制到多个 Broker 上，确保在单点故障下数据安全。
• ZooKeeper（或 KRaft 模式，Kafka Raft）用作分布式协调，维护集群状态。

---

7. 可扩展性

• Kafka 集群可以动态扩展，通过增加 Broker 处理更多数据流量。
• 生产者和消费者均可独立扩展，满足不同场景下的需求。

---

8. 灵活的数据流处理

• Kafka Streams API 提供了实时流数据处理能力，允许开发者直接处理和转换流数据。
• KSQL（基于 SQL 的流处理）为非技术用户提供了易用的数据处理能力。

---

9. 多种客户端支持

• Kafka 提供对多种编程语言的客户端支持，包括 Java、Python、Go、C++ 等。
• 生产者和消费者可以选择适合自己语言和框架的实现。

---

10. 强大的生态系统

• Kafka 提供多种工具扩展功能：
  • Kafka Connect：用于快速连接和集成外部系统（如数据库、文件系统、NoSQL 数据库）。
  • Schema Registry：管理消息的模式（Schema），确保数据格式一致性。
  • MirrorMaker：用于跨数据中心或跨集群的消息复制。

---

11. 多租户支持

• Kafka 通过隔离 Topic、访问控制和分区管理，支持多租户部署。
• 配合 ACL（访问控制列表）和 SSL/TLS，可以实现精细化的权限控制。

---

12. 一次性和至少一次交付语义

• 默认提供 "至少一次" 的消息交付语义。
• 可通过额外配置实现 "恰好一次" 的语义（例如，开启 idempotent producer 和事务）。

1.2. Kafka 的特性

Kafka 的设计目标：

• 高性能
  • 分区、分段、索引：基于分区机制提供并发处理能力。分段、索引提升了数据读写的查询效率。
  • 顺序读写：使用顺序读写提升磁盘 IO 性能。
  • 零拷贝：利用零拷贝技术，提升网络 I/O 效率。
  • 页缓存：利用操作系统的 PageCache 来缓存数据（典型的利用空间换时间）
  • 批量读写：批量读写可以有效提升网络 I/O 效率。
  • 数据压缩：Kafka 支持数据压缩，可以有效提升网络 I/O 效率。
  • pull 模式：Kafka 架构基于 pull 模式，可以自主控制消费策略，提升传输效率。
• 高可用
  • 持久化：Kafka 所有的消息都存储在磁盘，天然支持持久化。
  • 副本机制：Kafka 的 Broker 集群支持副本机制，可以通过冗余，来保证其整体的可用性。
  • 选举 Leader：Kafka 基于 ZooKeeper 支持选举 Leader，实现了故障转移能力。
• 伸缩性
  • 分区：Kafka 的分区机制使得其具有良好的伸缩性。

详细解释如下：

1. 高吞吐量和低延迟

• 目标: 支持每秒处理数百万条消息，且保持较低的处理延迟。
• 实现:
  • 使用顺序写入和零拷贝技术优化磁盘 I/O 性能。
  • 设计上避免消息中间路由或转换，直接传输生产者写入的数据。

---

2. 可扩展性

• 目标: 能够轻松扩展以支持大型分布式系统，从而处理不断增长的数据流量。
• 实现:
  • 通过分区（Partition）机制将数据流分布到多个节点（Broker）中。
  • 提供线性扩展能力，支持动态增加节点来提升吞吐量。

---

3. 高可用性

• 目标: 即使在单点故障发生时，系统也能持续提供服务。
• 实现:
  • 多副本机制（Replication）保证分区的数据在多个节点上存储。
  • 自动故障转移机制，确保 Leader 分区不可用时快速选举新的 Leader。

---

4. 持久性和可靠性

• 目标: 确保数据存储安全，并在需要时可以可靠地重新读取。
• 实现:
  • 数据写入磁盘后才标记为成功（日志存储）。
  • 提供可配置的日志保留策略（基于时间或存储空间）。

---

5. 顺序性保证

• 目标: 在单个分区中，保证消息以写入的顺序消费。
• 实现:
  • 每个分区在写入时按顺序存储消息。
  • 消费者从分区中顺序读取消息。

---

6. 灵活的发布-订阅模式

• 目标: 支持传统队列模型（点对点）和发布-订阅模型，满足不同的消息流需求。
• 实现:
  • 消费者组（Consumer Group）提供灵活的消息分发策略：
    • 点对点: 单个消费者组的多个消费者共享消息。
    • 广播: 不同消费者组独立消费同一条消息。

---

7. 与大数据生态系统的无缝集成

• 目标: 作为现代大数据处理的基础组件，与主流工具协同工作。
• 实现:
  • 内置 Kafka Connect 支持与外部系统（如 Hadoop、Spark、Elasticsearch）的高效集成。
  • 提供 Kafka Streams 和 KSQL，为实时流处理提供支持。

---

8. 多语言支持

• 目标: 支持多种编程语言，方便开发者使用。
• 实现:
  • 提供官方和社区维护的客户端库，支持 Java、Python、Go、C++、.NET 等。

---

9. 灵活的存储和消费模型

• 目标: 提供可定制的数据存储和消费方式。
• 实现:
  • 可配置的日志保留时间，允许长时间存储数据。
  • 消费者根据偏移量（Offset）任意位置读取消息。

---

10. 高度可靠的交付语义

• 目标: 支持多种消息交付语义（至少一次、至多一次、恰好一次）。
• 实现:
  • 至少一次: 默认语义，通过偏移量提交实现。
  • 恰好一次: 通过幂等生产者和事务支持实现。

---

11. 简化的运维和部署

• 目标: 降低分布式系统的管理复杂度。
• 实现:
  • 通过 ZooKeeper 或 KRaft 实现集群协调。
  • 提供丰富的监控指标和工具（如 JMX、Prometheus）。

---

12. 安全性

• 目标: 确保数据在传输和存储过程中的安全。
• 实现:
  • 支持 SSL/TLS 加密、SASL 验证机制。
  • 提供基于 ACL（访问控制列表）的精细化权限管理。

1.3. Kafka 术语

• 消息：Kafka 的数据单元被称为消息。消息由字节数组组成。
• 批次：批次就是一组消息，这些消息属于同一个主题和分区。
• 主题（Topic）：Kafka 消息通过主题进行分类。主题就类似数据库的表。
  • 不同主题的消息是物理隔离的；
  • 同一个主题的消息保存在一个或多个 Broker 上。但用户只需指定消息的 Topic 即可生产或消费数据而不必关心数据存于何处。
  • 主题有一个或多个分区。
• 分区（Partition）：分区是一个有序不变的消息序列，消息以追加的方式写入分区，然后以先入先出的顺序读取。Kafka 通过分区来实现数据冗余和伸缩性。
• 消息偏移量（Offset）：表示分区中每条消息的位置信息，是一个单调递增且不变的值。
• 生产者（Producer）：生产者是向主题发布新消息的 Kafka 客户端。生产者可以将数据发布到所选择的主题中。生产者负责将记录分配到主题中的哪一个分区中。
• 消费者（Consumer）：消费者是从主题订阅新消息的 Kafka 客户端。消费者通过检查消息的偏移量来区分消息是否已读。
• 消费者群组（Consumer Group）：多个消费者共同构成的一个群组，同时消费多个分区以实现高并发。
  • 每个消费者属于一个特定的消费者群组（可以为每个消费者指定消费者群组，若不指定，则属于默认的群组）。
  • 群组中，一个消费者可以消费多个分区
  • 群组中，每个分区只能被指定给一个消费
• 再均衡（Rebalance）：消费者组内某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程。分区再均衡是 Kafka 消费者端实现高可用的重要手段。
• Broker - 一个独立的 Kafka 服务器被称为 Broker。Broker 接受来自生产者的消息，为消息设置偏移量，并提交消息到磁盘保存；消费者向 Broker 请求消息，Broker 负责返回已提交的消息。
• 副本（Replica）：Kafka 中同一条消息能够被拷贝到多个地方以提供数据冗余，这些地方就是所谓的副本。副本还分为领导者副本和追随者副本，各自有不同的角色划分。副本是在分区层级下的，即每个分区可配置多个副本实现高可用。

1.4. Kafka 发行版本

1. Apache Kafka

• 维护者: Apache Software Foundation
• 特点:
  • 官方版本，完全开源（Apache 2.0 许可证）。
  • 更新频繁，功能较新，但相对来说需要更多配置和调优。
  • 适合希望自己管理基础设施的团队。
• 获取方式:
  • Apache Kafka 官网

---

2. Confluent Kafka

• 维护者: Confluent, Inc.（由 Kafka 的创始人创建）
• 特点:
  • 提供 Kafka 的增强版，支持更多企业级功能。
  • 包括 Schema Registry、Kafka Connect 插件、KSQL（流处理）、多集群管理等工具。
  • 有开源版本和商业化版本（支持 SLA、企业支持等）。
  • 提供 Confluent Cloud，完全托管的云端 Kafka 服务。
• 获取方式:
  • Confluent 官网

---

3. Amazon Managed Streaming for Kafka (MSK)

• 维护者: Amazon Web Services (AWS)
• 特点:
  • 提供完全托管的 Kafka 服务。
  • 简化了集群管理、扩展和维护工作。
  • 与 AWS 生态系统（如 S3、Redshift、Lambda）深度集成。
• 获取方式:
  • AWS MSK 服务

---

4. Azure Event Hubs for Kafka

• 维护者: Microsoft Azure
• 特点:
  • 提供 Kafka 协议兼容的托管服务。
  • 无需运行实际的 Kafka 集群，应用程序可以直接使用 Kafka API。
  • 与 Azure 的其他服务（如 Azure Blob Storage、Azure Functions）紧密集成。
• 获取方式:
  • Azure Event Hubs

---

5. Google Cloud Pub/Sub Lite

• 维护者: Google Cloud
• 特点:
  • 提供 Kafka API 的兼容接口。
  • 提供分布式消息队列的托管服务，支持动态扩展和数据备份。
  • 适合 Google Cloud 平台的用户。
• 获取方式:
  • Google Cloud Pub/Sub Lite

---

6. Red Hat AMQ Streams

• 维护者: Red Hat
• 特点:
  • 基于 Apache Kafka，提供了 OpenShift 和 Kubernetes 的深度集成。
  • 提供企业级支持和工具，例如监控和调优工具。
• 获取方式:
  • Red Hat AMQ Streams

---

7. Cloudera DataFlow (CDF) for Kafka

• 维护者: Cloudera
• 特点:
  • 提供 Kafka 及其相关工具的企业级支持。
  • 集成了大数据平台（如 Hadoop）和其他 Cloudera 产品。
  • 提供数据流可视化和监控工具。
• 获取方式:
  • Cloudera 官网

---

8. Strimzi Kafka Operator

• 维护者: Strimzi 社区（CNCF 项目）
• 特点:
  • 主要面向 Kubernetes 环境的 Kafka 部署和管理。
  • 提供 Operator 模式，实现 Kafka 的自动化部署和更新。
  • 完全开源。
• 获取方式:
  • Strimzi 官网

---

如何选择合适的版本？

• 自行管理 vs 托管服务: 如果想要灵活性和完全控制权，选择 Apache Kafka 或 Strimzi。如果希望减少运维工作量，可以选择托管服务（如 AWS MSK 或 Confluent Cloud）。
• 功能需求: 需要企业级功能或增强支持时，可选择 Confluent Kafka、Red Hat AMQ Streams 或 Cloudera CDF。
• 云集成: 如果已经使用特定云平台（AWS、Azure、Google），优先考虑对应的托管服务。

1.5. Kafka 重大版本

Apache Kafka 的版本号采用“主版本.次版本.修订版本”的格式。其中，主版本的变更通常代表重大更新，可能包含不兼容的 API 变更、新特性或架构调整。截至 2024 年 11 月，Kafka 的主要版本如下：

1. 0.x 系列：初始开发阶段，功能逐步完善。
2. 1.x 系列：引入了更稳定的特性和性能改进。
3. 2.x 系列：增加了多项新功能，如 Kafka Streams、改进的控制器机制等。
4. 3.x 系列：进一步增强了性能和稳定性，改进了元数据管理和集群扩展能力。

每个主版本下的次版本和修订版本通常包含功能改进、性能优化和漏洞修复。在升级 Kafka 时，建议仔细阅读官方发布的版本说明，以了解新版本的特性、改进和可能的兼容性问题。

Kafka 0.x

• 初始版本: 项目起步，主要功能框架确立。
• 特点:
  • 引入分区和副本机制。
  • 基于 zookeeper 的分布式协调。
  • 使用简单的日志存储。

---

Kafka 1.x (2017 年发布)

• 意义: 第一次标志性稳定版本。
• 新功能:
  • 生产者和消费者协议的稳定性增强。
  • Kafka Streams: 增加了流处理 API，支持实时流数据处理。
  • 更好的兼容性和更长的稳定支持周期。
• 注意: 1.x 版本与后续版本向后兼容性较好。

---

Kafka 2.x (2018-2021)

• 改进的关键点:
  • 新 API:
    • 提供了更多的 Streams API 功能。
    • 增加了 AdminClient，支持程序化管理 Kafka 集群。
  • 协议改进:
    • 增强 Kafka 消息格式和分区副本协议。
    • 支持 Zstd 压缩。
  • 安全性:
    • 加强了对 ACL（访问控制列表）的支持。
  • 性能优化:
    • 提升了生产者、消费者和 broker 的吞吐量。
    • 增强了 Controller 的性能，特别是在分区重分配期间。
  • Streams API: 支持全局状态存储。

---

Kafka 3.x (2021 至今)

• 重大变更:
  • 移除旧版功能:
    • 移除过时的 Scala Producer 和 Consumer API。
    • 逐步淘汰依赖于 Zookeeper 的实现。
  • KRaft 模式（Kafka Raft Metadata）:
    • 引入基于 Raft 协议的新的元数据管理模式，减少对 Zookeeper 的依赖。
    • 在部分场景下完全移除 Zookeeper。
  • 动态配置:
    • 支持更多 Broker 和 Topic 的动态更新配置。
  • 性能改进:
    • 更高效的存储和日志清理机制。
    • 改善延迟问题。
  • 生态支持:
    • 对 Kubernetes 和云环境的支持更加成熟。
  • 新的功能:
    • 增加自定义分区器。
    • 消息压缩支持更多算法。
• 向前展望:
  • Zookeeper 的完全替代。
  • 更轻量化的 Broker 管理。

如何选择 Kafka 的重大版本？

1. 生产稳定性:

   • 如果需要稳定性和社区成熟支持，可以选择最新的稳定版本（如 3.x）。
   • 如果仍然依赖 Zookeeper，建议使用 2.x 或支持 KRaft 的 3.x。

2. 功能需求:

   • 如果需要最新特性（如 KRaft 模式或高级 Streams API），请选择 3.x。
   • 如果功能需求简单，1.x 和 2.x 仍然可以满足。

3. 生态支持:

   • 如果运行在 Kubernetes 上，3.x 版本的改进适合现代云原生环境。

2. Kafka 服务端使用入门

2.1. 步骤一、获取 Kafka

下载最新的 Kafka 版本并解压到本地。

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

2.2. 步骤二、启动 Kafka 环境

这里说明下，我不会再使用ZooKeeper了，因为kafka后期会完全移除ZooKeeper了。所以本地我使用KRaft启动，本地环境为mac。

生成群集UUID

$ KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

格式化日志目录

$ bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/kraft/reconfig-server.properties

启动server

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

一旦Kafka服务器成功启动，您将有一个基本的Kafka环境运行并准备使用。

启动后会提示启动成功

注意：您的本地环境必须安装Java 8+。

Apache Kafka可以使用KRaft或ZooKeeper启动。要开始使用任何一种配置，请按照以下部分之一进行操作，但不要同时进行。

2.3. 步骤三、创建一个 TOPIC 并存储您的事件

Kafka 是一个分布式事件流处理平台，它可以让您通过各种机制读、写、存储并处理事件（events (opens new window)，也被称为记录或消息）

示例事件包括付款交易，手机的地理位置更新，运输订单，物联网设备或医疗设备的传感器测量等等。 这些事件被组织并存储在主题中（topics (opens new window)）。 简单来说，主题类似于文件系统中的文件夹，而事件是该文件夹中的文件。

因此，在您写入第一个事件之前，您必须先创建一个 Topic。执行以下指令：

bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

所有的 Kafka 命令行工具都有附加可选项：不加任何参数，运行 kafka-topics.sh 命令会显示使用信息。例如，会显示新 Topic 的分区数等细节。

$ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
Topic:quickstart-events  PartitionCount:1    ReplicationFactor:1 Configs:
    Topic: quickstart-events Partition: 0    Leader: 0   Replicas: 0 Isr: 0

2.4. 步骤四、向 Topic 写入 Event

Kafka 客户端和 Kafka Broker 的通信是通过网络读写 Event。一旦收到信息，Broker 会将其以您需要的时间（甚至永久化）、容错化的方式存储。

执行 kafka-console-producer.sh 命令将 Event 写入 Topic。默认，您输入的任意行会作为独立 Event 写入 Topic：

$ bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
This is my first event
This is my second event

您可以通过 Ctrl-C 在任何时候中断 kafka-console-producer.sh

2.5. 步骤五、读 Event

执行 kafka-console-consumer.sh 以读取写入 Topic 中的 Event

$ bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event

您可以通过 Ctrl-C 在任何时候中断 kafka-console-consumer.sh

由于 Event 被持久化存储在 Kafka 中，因此您可以根据需要任意多次地读取它们。 您可以通过打开另一个终端会话并再次重新运行上一个命令来轻松地验证这一点。

2.6. 步骤六、通过 KAFKA CONNECT 将数据作为事件流导入/导出

您可能有大量数据，存储在传统的关系数据库或消息队列系统中，并且有许多使用这些系统的应用程序。 通过 Kafka Connect (opens new window)，您可以将来自外部系统的数据持续地导入到 Kafka 中，反之亦然。 因此，将已有系统与 Kafka 集成非常容易。为了使此过程更加容易，有数百种此类连接器可供使用。

需要了解有关如何将数据导入和导出 Kafka 的更多信息，可以参考：Kafka Connect section (opens new window)章节

2.7. 步骤七、使用 Kafka Streams 处理事件

一旦将数据作为 Event 存储在 Kafka 中，就可以使用 Kafka Streams (opens new window)的 Java / Scala 客户端。它允许您实现关键任务的实时应用程序和微服务，其中输入（和/或）输出数据存储在 Kafka Topic 中。

Kafka Streams 结合了 Kafka 客户端编写和部署标准 Java 和 Scala 应用程序的简便性，以及 Kafka 服务器集群技术的优势，使这些应用程序具有高度的可伸缩性、弹性、容错性和分布式。该库支持一次性处理，有状态的操作，以及聚合、窗口化化操作、join、基于事件时间的处理等等。

KStream<String, String> textLines = builder.stream("quickstart-events");

KTable<String, Long> wordCounts = textLines
            .flatMapValues(line -> Arrays.asList(line.toLowerCase().split(" ")))
            .groupBy((keyIgnored, word) -> word)
            .count();

wordCounts.toStream().to("output-topic"), Produced.with(Serdes.String(), Serdes.Long()));

Kafka Streams demo (opens new window)和 app development tutorial (opens new window)展示了如何从头到尾的编码并运行一个流式应用。

2.8. 步骤八、终止 Kafka 环境

1. 如果尚未停止，请使用 Ctrl-C 停止生产者和消费者客户端。
2. 使用 Ctrl-C 停止 Kafka服务。
3. 最后，使用 Ctrl-C 停止 ZooKeeper 服务器。

如果您还想删除本地 Kafka 环境的所有数据，包括您在此过程中创建的所有事件，请执行以下命令：

$ rm -rf /tmp/kafka-logs /tmp/zookeeper

当然了，我们没有ZooKeeper，可以忽略第三步。

 

已经成功关闭了kafka server。 

3. Kafka Java 客户端使用入门

3.1. 引入 maven 依赖

Stream API 的 maven 依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>3.9.0</version>
</dependency>

其他 API 的 maven 依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.9.0</version>
</dependency>

3.2. Kafka 核心 API

Kafka 提供了多个核心 API，它们覆盖了消息生产、消费、处理和连接外部系统的功能。这些 API 是 Kafka 实现其分布式流处理平台的重要组成部分。

---

1. Producer API

• 作用: 用于向 Kafka 中的指定主题（Topic）发送消息。
• 功能:
  • 支持同步或异步方式发送消息。
  • 可指定分区策略（例如按键分区、随机分区）。
  • 支持幂等生产者（Idempotent Producer），确保消息不重复发送。
• 应用场景: 数据采集、实时日志记录等。

---

2. Consumer API

• 作用: 用于从 Kafka 的主题中读取消息。
• 功能:
  • 支持消息的顺序消费（按分区）。
  • 提供消费者组（Consumer Group）机制，实现负载均衡和多消费者协作。
  • 允许手动或自动提交偏移量（Offset），控制消息的读取位置。
• 应用场景: 日志分析、事件驱动应用等。

---

3. Streams API

• 作用: 用于构建实时流处理应用。
• 功能:
  • 支持状态存储，能够处理有状态的流数据。
  • 提供高级操作，例如过滤、聚合、联结（Join）。
  • 能够从一个或多个主题读取数据，实时处理后写回 Kafka。
• 应用场景: 数据清洗、实时监控、复杂事件处理等。

---

4. Connect API

• 作用: 用于集成 Kafka 和外部数据源或目标。
• 功能:
  • 通过 Source Connectors 从外部系统导入数据到 Kafka。
  • 通过 Sink Connectors 将 Kafka 数据导出到外部系统。
  • 提供分布式运行模式和任务管理功能，简化数据管道的配置。
• 应用场景: 数据同步、ETL 流程。

---

5. Admin API

• 作用: 用于管理和操作 Kafka 的集群和主题。
• 功能:
  • 创建、删除、列出主题。
  • 修改主题配置，例如分区数、副本因子。
  • 查询集群状态（Broker、分区、ISR 等）。
• 应用场景: Kafka 运维和自动化脚本开发。

---

API 功能对比

API	主要功能	常见场景
Producer API	生产消息并发送到主题	数据采集、事件发布
Consumer API	从主题读取消息	数据消费、事件驱动架构
Streams API	实时处理和转换数据流	数据清洗、实时分析
Connect API	与外部系统集成，实现数据流的输入和输出	数据同步、ETL 流程
Admin API	管理主题、分区、副本等集群资源	自动化运维、配置管理

---

API 集成与生态

• Schema Registry: 搭配 Confluent 提供的 Schema Registry，可以管理和验证消息格式（Schema）。
• Kafka Tools: Kafka 的 CLI 工具可以与 API 功能结合，用于测试和运维。

3.3. 发送消息

发送并忽略返回

代码如下，直接通过 send 方法来发送

ProducerRecord<String, String> record =
            new ProducerRecord<>("CustomerCountry", "Precision Products", "France");
    try {
            producer.send(record);
    } catch (Exception e) {
            e.printStackTrace();
}

同步发送

代码如下，与“发送并忘记”的方式区别在于多了一个 get 方法，会一直阻塞等待 Broker 返回结果：

ProducerRecord<String, String> record =
            new ProducerRecord<>("CustomerCountry", "Precision Products", "France");
    try {
            producer.send(record).get();
    } catch (Exception e) {
            e.printStackTrace();
}

异步发送

代码如下，异步方式相对于“发送并忽略返回”的方式的不同在于：在异步返回时可以执行一些操作，如记录错误或者成功日志。

首先，定义一个 callback

private class DemoProducerCallback implements Callback {
      @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
           if (e != null) {
               e.printStackTrace();
             }
        }
}

然后，使用这个 callback

ProducerRecord<String, String> record =
            new ProducerRecord<>("CustomerCountry", "Biomedical Materials", "USA");
producer.send(record, new DemoProducerCallback());

发送消息示例

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * Kafka 生产者生产消息示例 生产者配置参考：https://kafka.apache.org/documentation/#producerconfigs
 */
public class ProducerDemo {
    private static final String HOST = "localhost:9092";

    public static void main(String[] args) {
        // 1. 指定生产者的配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, HOST);
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        properties.put(ProducerConfig.RETRIES_CONFIG, 0);
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
            "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            "org.apache.kafka.common.serialization.StringSerializer");

        // 2. 使用配置初始化 Kafka 生产者
        Producer<String, String> producer = new KafkaProducer<>(properties);

        try {
            // 3. 使用 send 方法发送异步消息
            for (int i = 0; i < 100; i++) {
                String msg = "Message " + i;
                producer.send(new ProducerRecord<>("HelloWorld", msg));
                System.out.println("Sent:" + msg);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 4. 关闭生产者
            producer.close();
        }
    }
}

3.4. 消费消息流程

消费流程

具体步骤如下：

1. 创建消费者。
2. 订阅主题。除了订阅主题方式外还有使用指定分组的模式，但是常用方式都是订阅主题方式
3. 轮询消息。通过 poll 方法轮询。
4. 关闭消费者。在不用消费者之后，会执行 close 操作。close 操作会关闭 socket，并触发当前消费者群组的再均衡。

    // 1.构建KafkaCustomer
    Consumer consumer = buildCustomer();

    // 2.设置主题
    consumer.subscribe(Arrays.asList(topic));

    // 3.接受消息
    try {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(500);
            System.out.println("customer Message---");
            for (ConsumerRecord<String, String> record : records)

                // print the offset,key and value for the consumer records.
                System.out.printf("offset = %d, key = %s, value = %s\n",
                        record.offset(), record.key(), record.value());
        }
    } finally {
        // 4.关闭消息
            consumer.close();
    }

创建消费者的代码如下：

public Consumer buildCustomer() {
    Properties props = new Properties();
    // bootstrap.servers是Kafka集群的IP地址。多个时,使用逗号隔开
    props.put("bootstrap.servers", "localhost:9092");
    // 消费者群组
    props.put("group.id", "test");
    props.put("enable.auto.commit", "true");
    props.put("auto.commit.interval.ms", "1000");
    props.put("session.timeout.ms", "30000");
    props.put("key.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
    KafkaConsumer<String, String> consumer = new KafkaConsumer
            <String, String>(props);

    return consumer;
}

消费消息方式

分为订阅主题和指定分组两种方式：

• 消费者分组模式。通过订阅主题方式时，消费者必须加入到消费者群组中，即消费者必须有一个自己的分组；
• 独立消费者模式。这种模式就是消费者是独立的不属于任何消费者分组，自己指定消费那些 Partition。

（1）订阅主题方式

consumer.subscribe(Arrays.asList(topic));

（2）独立消费者模式

通过 consumer 的 assign(Collection<TopicPartition> partitions) 方法来为消费者指定分区。

public void consumeMessageForIndependentConsumer(String topic){
    // 1.构建KafkaCustomer
    Consumer consumer = buildCustomer();

    // 2.指定分区
    // 2.1获取可用分区
    List<PartitionInfo> partitionInfoList = buildCustomer().partitionsFor(topic);
    // 2.2指定分区,这里是指定了所有分区,也可以指定个别的分区
    if(null != partitionInfoList){
        List<TopicPartition> partitions = Lists.newArrayList();
        for(PartitionInfo partitionInfo : partitionInfoList){
            partitions.add(new TopicPartition(partitionInfo.topic(),partitionInfo.partition()));
        }
        consumer.assign(partitions);
    }

    // 3.接受消息
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(500);
        System.out.println("consume Message---");
        for (ConsumerRecord<String, String> record : records) {

            // print the offset,key and value for the consumer records.
            System.out.printf("offset = %d, key = %s, value = %s\n",
                    record.offset(), record.key(), record.value());

            // 异步提交
            consumer.commitAsync();


        }
    }
}

完整的消费代码：

package onyx;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class ConsumerDemo {
    public static void main(String[] args) {
        // 1.构建KafkaCustomer
        Consumer consumer = buildCustomer();

        // 2.设置主题
        consumer.subscribe(Arrays.asList("HelloWorld"));

        // 3.接受消息
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(500);
                System.out.println("customer Message---");
                for (ConsumerRecord<String, String> record : records) {
                    // print the offset,key and value for the consumer records.
                    System.out.printf("offset = %d, key = %s, value = %s\n",
                            record.offset(), record.key(), record.value());
                }
            }
        } finally {
            // 4.关闭消息
            consumer.close();
        }
    }

    public static Consumer buildCustomer() {
        Properties props = new Properties();
        // bootstrap.servers是Kafka集群的IP地址。多个时,使用逗号隔开
        props.put("bootstrap.servers", "localhost:9092");
        // 消费者群组
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer
                <String, String>(props);

        return consumer;
    }

}

 参考资料：

Kafka 官网

Kafka 快速入门 | BIGDATA-TUTORIAL

原文地址：https://blog.csdn.net/u010398771/article/details/143903777

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！