点一下关注吧！！！非常感谢！！持续更新！！！

🚀 AI 篇持续更新中！（长期更新）

AI 炼丹日志-31- 千呼万唤始出来 GPT-5 发布！“快的模型 + 深度思考模型 + 实时路由”，持续打造实用 AI 工具指南！📐🤖

💻 Java 篇正式开启！（300 篇）

目前 2025 年 08 月 18 日更新到：Java-100 深入浅出 MySQL 事务隔离级别：读未提交、已提交、可重复读与串行化 MyBatis 已完结，Spring 已完结，Nginx 已完结，Tomcat 已完结，分布式服务正在更新！深入浅出助你打牢基础！

📊 大数据板块已完成多项干货更新（300 篇）：

包括 Hadoop、Hive、Kafka、Flink、ClickHouse、Elasticsearch 等二十余项核心组件，覆盖离线+实时数仓全栈！大数据-278 Spark MLib - 基础介绍 机器学习算法 梯度提升树 GBDT 案例 详解

章节内容

上节我们完成了如下的内容：

• Kafka 延时队列

• Kafka 重试队列

• Kafka JavaAPI 实现 重试队列的操作

应用场景

消息传递

Kafka 可以很好的替代传统的邮件代理，消息代理的使用有很多种原因（将处理与数据生产者分离，缓冲未处理消息等）。与大多数邮件系统相比，Kafka 具有更好的吞吐量，内置的分区，复制和容错功能，这使其成为大规模邮件处理应用程序的理想解决方案。

网站活动路由

Kafka 最初的用例是能够将用户活动跟踪管道重建为一组实时的发布-订阅。这意味着将网站活动（页面浏览、搜索、其他操作等）发布到主题中心，每种活动类型只有一个主题。这些可用于一系列的用例的订阅，包括实时处理，实时监控，以及加载到 Hadoop 或脱机数据仓库系统中以进行脱机处理和报告。活动跟踪通常量很大，因为每个用户页面视图都会生成许多活动消息。

监控指标

Kafka 通常用于操作监控数据，这涉及汇总来自分布式应用程序的统计信息，以生成操作数据的集中。

日志汇总

许多人使用 Kafka 代替日志聚合解决方案，日志聚合通常从服务器收集物理日志文件，并将它们放在中央位置（也许是文件服务器或 HDFS）以进行处理。Kafka 提取文件的详细信息，并以日志的形式更清晰的抽象日志或事件数据，这允许较低的延迟的处理，并更容易支持多个数据源和分布式数据消耗。以日志为中心的系统（例如 Scribe 或 Flume）相比，Kafka 具有同样出色的性能，由于复制而提供的更强的耐用性保证以及更低的端到端的延迟。

流处理

Kafka 提供了一个强大的流处理架构，允许用户在由多个阶段组成的处理管道中高效地处理数据。这种管道式处理方式通常包括以下几个关键步骤：

1. 数据摄入阶段：原始输入数据从 Kafka 主题中被消费。例如，在新闻推荐系统中，这个阶段可能从 RSS 提要中检索文章内容，并将其发布到名为"raw-articles"的主题中。

2. 数据处理阶段：数据会经过一系列转换操作：

3. 数据清洗：对内容进行规范化处理（如统一编码格式、去除 HTML 标签）

4. 数据去重：使用消息键或内容哈希来识别和过滤重复文章

5. 内容增强：可能添加元数据或进行情感分析处理后的数据会被发布到新的主题，如"cleaned-articles"

6. 业务逻辑阶段：实现核心业务功能。在新闻推荐场景中，这个阶段可能：

7. 分析用户阅读历史

8. 计算文章相似度

9. 应用推荐算法最终结果会被发布到"recommendations"主题供下游系统使用

从 Kafka 0.10.0.0 版本开始，引入了 Kafka Streams 这个轻量级但功能强大的流处理库。Kafka Streams 的主要特点包括：

• 与 Kafka 深度集成：直接利用 Kafka 的分布式架构和分区机制

• Exactly-once 语义：确保数据处理不会重复或丢失

• 状态管理：支持本地状态存储，便于实现窗口聚合等操作

典型的 Kafka Streams 应用场景包括：

• 实时监控和告警系统

• 实时分析仪表板

• 复杂事件处理

• 微服务间的数据转换

除了 Kafka Streams，Kafka 生态中还有其他流处理工具：

1. Apache Storm：

2. 最早的实时计算系统之一

3. 使用 spout 和 bolt 的处理模型

4. 适合要求低延迟的场景

5. Apache Samza：

6. 与 YARN 深度集成

7. 提供可插拔的存储后端

8. 适合与 Hadoop 生态系统集成的场景

这些工具都基于 Kafka 主题创建实时数据流处理图，但各有侧重。选择时需要考虑：

• 开发复杂度

• 延迟要求

• 与现有系统的集成度

• 运维成本等因素

活动采集

事件源是一种应用程序，其中状态更改以时间顺序记录记录。Kafka 对大量存储的日志数据的支持使其成为以这种样式构建的应用程序的绝佳后端。

提交日志

Kafka 可以用作分布式系统的一种外部提交日志，该日志有助于在节点之间复制数据，并充当故障节点恢复其数据的重新同步机制。Kafka 中的日志压缩功能有助于支持此用法。

集群搭建

集群设计

由于之前我们已经搭建过单机的 Kafka，同时我们为了做之前的实验，一共搭建了两台 Kafka 的小集群（用作 Broker 宕机之后的分区、副本等内容的测试），这里我们将对一些部分进行简化。

机器详情

目前我们有三台云服务：

• h121.wzk.icu

• h122.wzk.icu

• h123.wzk.icu

我们已经搭建好了，ZooKeeper 的集群，如果你还没有搭建，需要回到之前的章节：ZooKeeper 集群搭建。这里开始，我们直接搭建 Kafka 的集群环境。在 h121.wzk.icu 中，我我们已经有了：kafka_2.12-2.7.2 且是配置好的。

我们借助之前 Hadoop 中编写的 Shell 工具来完成 Kafka 文件的分发（你也可以使用别的方法，比如压缩包等等）

rsync-script kafka_2.12-2.7.2/

h121

h122

h123

环境变量

我们在三台节点上，尽量配置好环境变量：

• JDK

• ZooKeeper

• Kafka

修改配置

h121

vim /opt/servers/kafka_2.12-2.7.2/config/server.properties

修改如下内容：

# h121是0、h122是1、h123是2broker.id=0
# 数据的存储目录log.dirs=/opt/kafka-logs
# ZooKeeper 地址zookeeper.connect=h121.wzk.icu:2181,h122.wzk.icu:2181,h123.wzk.icu:2181

对应的内容截图如下所示：

h122

# h121是0、h122是1、h123是2broker.id=1
# 数据的存储目录log.dirs=/opt/kafka-logs
# ZooKeeper 地址zookeeper.connect=h121.wzk.icu:2181,h122.wzk.icu:2181,h123.wzk.icu:2181

h123

# h121是0、h122是1、h123是2broker.id=3
# 数据的存储目录log.dirs=/opt/kafka-logs
# ZooKeeper 地址zookeeper.connect=h121.wzk.icu:2181,h122.wzk.icu:2181,h123.wzk.icu:2181

对应的截图如下图所示：

启动集群

在每台节点上都执行：

kafka-server-start.sh /opt/servers/kafka_2.12-2.7.2/config/server.properties

查看集群

我们需要进入 ZooKeeper 来启动服务：

zkCli.sh# 进入ZK后，我们查看当中的信息ls /brokers/ids

执行结果如下图所示：第一次执行的时候，我的第三台没有配置好环境变量，启动失败了，第二次可以看到：【0,1,2】

h121

[zk: localhost:2181(CONNECTED) 2] get /brokers/ids/0{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://h121.wzk.icu:9092"],"jmx_port":-1,"features":{},"host":"h121.wzk.icu","timestamp":"1722931444209","port":9092,"version":5}[zk: localhost:2181(CONNECTED) 3] 

h122

[zk: localhost:2181(CONNECTED) 3] get /brokers/ids/1{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://h122.wzk.icu:9092"],"jmx_port":-1,"features":{},"host":"h122.wzk.icu","timestamp":"1722931447944","port":9092,"version":5}

h123

[zk: localhost:2181(CONNECTED) 4] get /brokers/ids/2{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://h123.wzk.icu:9092"],"jmx_port":-1,"features":{},"host":"h123.wzk.icu","timestamp":"1722931584357","port":9092,"version":5}

如下所示：