文章目录
- 一、前言
- 二、环境准备
- 三、环境搭建
- 3.1 JDK 环境搭建
- 3.2 Scala 环境搭建
- 3.2.1 配置 Scala 环境变量
- 3.2.2 验证
- 3.3 Gradle 环境搭建
- 3.3.1 配置 Gradle 环境变量
- 3.3.2 验证
- 3.4 Zookeeper 环境搭建
- 3.4.1 配置 Zookeeper 环境变量
- 3.4.2 验证
- 3.5 Kafka 源码搭建
- 3.5.1 导入 Kafka 源码至 IDEA
- 3.5.2 修改 build.gradle 配置
- 四、源码代码结构
- 五、核心 Core 目录代码结构
- 六、源码部署验证
- 6.1 core.main 下新建日志配置类
- 6.2 修改 server.properties
- 6.3 配置 Run Configurations
- 6.4 发送、消费 message
一、前言
参考的博客是用的是 Kafka 2.7 版本的,于是我这里也采用 2.7 版本,理由是这个版本还保存着 zookeeper,生产环境比较稳定
二、环境准备
- JDK:1.8.0_241
- Scala:2.12.8
- Gradle:6.6
- Zookeeper:3.4.14
三、环境搭建
3.1 JDK 环境搭建
直接使用本地环境
3.2 Scala 环境搭建
下载链接:Scala 2.12.8 | The Scala Programming Language (scala-lang.org)
3.2.1 配置 Scala 环境变量
vim ~/.bash_profile
# 安装的路径
export SCALA_HOME=/Users/gabriel/Environment/scala-2.12.8
export PATH=$PATH:$SCALA_HOME/bin
source ~/.bash_profile
3.2.2 验证
输入 scala -version 看到提示则成功
3.3 Gradle 环境搭建
下载链接:Gradle Distributions
下载版本为 gradle-6.6-bin
3.3.1 配置 Gradle 环境变量
vim ~/.bash_profile
# 安装的路径
export GRADLE_HOME=/Users/gabriel/Environment/gradle-6.6
export PATH=$PATH:$GRADLE_HOME/bin
source ~/.bash_profile
3.3.2 验证
输入 gradle -version 看到提示则成功
3.4 Zookeeper 环境搭建
下载链接:Index of /dist/zookeeper/zookeeper-3.4.14 (apache.org)
3.4.1 配置 Zookeeper 环境变量
vim ~/.bash_profile
# 安装的路径
export ZK_HOME=/Users/gabriel/Environment/zookeeper-3.4.14
export PATH=$PATH:$ZK_HOME/bin
source ~/.bash_profile
创建 data 文件夹用于存放日志数据,并在 cfg 文件中指定
cd zookeeper-3.4.14
mkdir data
cd conf
mv zoo_sample.cfg zoo.cfg
# 修改 zoo.cfg 内的配置
dataDir=/Users/gabriel/Environment/zookeeper-3.4.14/data
3.4.2 验证
sh zkServer.sh start
输出提示则成功
3.5 Kafka 源码搭建
下载链接:Apache Kafka
这里选择了 2.7.0 版本
3.5.1 导入 Kafka 源码至 IDEA
Setting 里面需要配置 Gradle Home 为本地地址 /Users/gabriel/Environment/gradle-6.6
3.5.2 修改 build.gradle 配置
这里是我第一次使用 gradle 构建项目,基本原理和 maven 差不多,刚好学习一下
需要把镜像文件下载服务器更换为国内的私服,否则会相当慢,直接导致 “time out” 报错。
进入 kafka 源码包,修改 build.gradle 文件,在原来配置上,添加 ali 私服配置。
buildscript {
repositories {
maven {
url 'https://maven.aliyun.com/nexus/content/groups/public/'
}
maven {
url 'https://maven.aliyun.com/nexus/content/repositories/jcenter'
}
// 插件下载地址
maven {
url 'https://maven.aliyun.com/repository/gradle-plugin'
}
}
}
allprojects {
repositories {
maven {
url 'https://maven.aliyun.com/nexus/content/groups/public/'
}
maven {
url 'https://maven.aliyun.com/nexus/content/repositories/jcenter'
}
// 插件下载地址
maven {
url 'https://maven.aliyun.com/repository/gradle-plugin'
}
}
}
Bug 1:
疯狂报错如下,好像是 gradle 的插件找不到,从对应的阿里云地址也下载不下来
解决方案:在上面配置 maven 下载地址的地方加上专门的插件下载地址
A problem occurred configuring root project 'kafka-2.7.0-src'.
> Could not resolve all artifacts for configuration ':classpath'.
> Could not find gradle.plugin.com.github.spotbugs.snom:spotbugs-gradle-plugin:4.4.4.
Searched in the following locations:
- https://maven.aliyun.com/nexus/content/groups/public/gradle/plugin/com/github/spotbugs/snom/spotbugs-gradle-plugin/4.4.4/spotbugs-gradle-plugin-4.4.4.pom
- https://maven.aliyun.com/nexus/content/repositories/jcenter/gradle/plugin/com/github/spotbugs/snom/spotbugs-gradle-plugin/4.4.4/spotbugs-gradle-plugin-4.4.4.pom
- https://dl.bintray.com/content/netflixoss/external-gradle-plugins/gradle/plugin/com/github/spotbugs/snom/spotbugs-gradle-plugin/4.4.4/spotbugs-gradle-plugin-4.4.4.pom
Required by:
project :
> Could not find org.gradle:test-retry-gradle-plugin:1.1.6.
Searched in the following locations:
- https://maven.aliyun.com/nexus/content/groups/public/org/gradle/test-retry-gradle-plugin/1.1.6/test-retry-gradle-plugin-1.1.6.pom
- https://maven.aliyun.com/nexus/content/repositories/jcenter/org/gradle/test-retry-gradle-plugin/1.1.6/test-retry-gradle-plugin-1.1.6.pom
- https://dl.bintray.com/content/netflixoss/external-gradle-plugins/org/gradle/test-retry-gradle-plugin/1.1.6/test-retry-gradle-plugin-1.1.6.pom
Required by:
project :
Possible solution:
- Declare repository providing the artifact, see the documentation at https://docs.gradle.org/current/userguide/declaring_repositories.html
然后直接点右上角的小图标下载依赖
依赖下载完以后需要编译源代码(源代码中可以看到编译的 build 目录),这样子后面部署 kafka.Kafka 启动类然后使用脚本创建 producer、consumer 的时候,才能成功。手动输入命令如下:
./gradlew clean build -x test
或者直接右侧 gradle build,不过其中需要经过很多 test 测试,上面手动跳过
四、源码代码结构
重点看下其中几个重要的模块:
-
bin 目录:保存 Kafka 工具脚本,包括 kafka-server-start 和 kafka-console-producer 等脚本都存放在这里
-
clients 目录:保存 Kafka 客户端代码,比如生产者和消费者的代码都在该目录下(对于 Kafka 而言都是客户端)
-
config 目录:保存 Kafka 的配置文件,其中比较重要的配置文件是 server.properties
-
connect 目录:保存 Connect 组件的源代码。 Kafka Connect 组件是用来实现 Kafka 与外部系统之间的实时数据传输的。
-
core 目录:保存 Broker 端代码。Kafka 服务器端代码全部保存在该目录下(这对于 Kafka 而言是服务端)
-
generator 目录:Kafka 消息类处理模块,主要是根据 clients 模块下的 message json 文件生成对应的 java 类,在 build.gradle 文件中,可以看到定义了一个任务 processMessages
-
raft 目录:raft 一致性协议相关。
五、核心 Core 目录代码结构
- admin 包:执行管理命令的功能;
- api 包:封装请求和响应 DTO 对象;
- cluster 包:集群对象,例如 Partition 类代表一个分区,Replica 类代表一个分区副本;
- common 包:通用类,比如异常类
- consumer 包:只有一个类 BaseConsumerRecord,后面会丢弃该包,用 clients 包下 org.apache.kafka.clients.consumer.ConsumerRecord 代替;
- controller 包: 和 kafkaController(kc)相关的类,重点模块,一个kafka集群只有一个leader kafkaController,该kafkaController负责分区管理,副本管理,并保证集群信息在集群中同步;
- coordinator 包:保存了消费者端的 GroupCoordinator 代码和用于事务的 TransactionCoordinator 代码。对 coordinator 包进行分析,特别是对消费者端的 GroupCoordinator 代码进行分析,是 Broker 端协调者组件设计原理的关键;
- log 包:保存了 Kafka 最核心的日志结构代码,包括日志、日志段、索引文件等, 另外,该包下还封装了 Log Compaction 的实现机制,是非常重要的源码包;
- network 包:封装了 Kafka 服务器端网络层的代码,特别是 SocketServer.scala 这个文件,是 Kafka 实现 Reactor 模式的具体操作类,非常值得一读;
- server 包:顾名思义,它是 Kafka 的服务器端主代码,里面的类非常多,很多关键的 Kafka 组件都存放在这里,比如状态机、Purgatory 延时机制等;
- tools 包:工具类。
六、源码部署验证
最终其实就是在第三章编译完 kafka 所有文件的基础上,要启动 core 目录下和 broker 相关的类,启动类就是 kafka-2.7.0-src/core/src/main/scala/kafka/Kafka.scala
6.1 core.main 下新建日志配置类
在 core/main 下新建 resources 目录,再将 config/log4j.properties 拷贝到该目录下
这是为了避免启动报错“SLF4J: Failed to load class “org.slf4j.impl.StaticLoggerBinder””
6.2 修改 server.properties
修改 server.properties 中的 log.dirs 值,变为本地目录,
/Users/gabriel/Desktop/DiveinCode/kafka-2.7.0-src 下新建 kafka-logs 目录,修改配置如下
log.dirs=/Users/gabriel/Desktop/DiveinCode/kafka-2.7.0-src/kafka-logs
6.3 配置 Run Configurations
6.4 发送、消费 message
上面已经搭建好 Kafka Broker 的部署环境,并且编译成功了所有和 producer、consumer 有关的文件类,现在使用 Kafka 自带的脚本工具测试 producer 和 consumer
整体流程如下:
-
创建 topic
进入 kafka-2.7.0-src/bin 通过 kafka-topics.sh 命令来创建一个名为 topicjxz 的 topic(前提代码已经 build 编译完),zookeeper 的地址我是先通过 sh zkServer.sh start 启动,然后 zkCli 启动客户端找到的:
sh kafka-topics.sh --zookeeper localhost:2181 --create --topic topicjxz --replication-factor 1 --partitions 1
- 通过 kafka-console-consumer.sh 命令启动一个命令行的 consumer 来消费 topicjxz 这个 topic
sh kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicjxz
- 重新启动一个命令行终端,通过 kafka-console-producer.sh 命令启动一个命令行的 producer 向 topicjxz 这个 topic 中生产数据
sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic topicjxz
至此,我们发现已经将整个 producer、broker、consumer 的链路部署起来并测试成功了。