前言
Kafka 是通过 Scala 和 Java 独特编写的语言,之所以抉择 2.7.2 的版本是因为这个版本的 Kafka 是最初一版本保留 ZK 的版本。
为什么不间接部署最新版代码?
因为过来很长一段时间 Kafka 都是和 ZK 配合的,并且有很多成熟我的项目都应用了带 ZK 的 Kafka,去 ZK 的 Kafka 还有不少的路要走。
环境筹备
- JDK:1.8.0_351
- Scala:2.12.8
- Gradle:6.6
- Zookeeper:3.4.14
Kafka 2.7.2
倡议 fork 一个官网的分支到本人的仓库,不便本人学习的时候增加正文等内容。
地址:https://github.com/apache/kafka
装置 JDK 1.8
JDK 的教程玩网上有十分多,寻找适合的 JDK8 的版本即可。
https://www.oracle.com/java/technologies/javase/javase8-archive-downloads.html
装置 Scala 2.12.8
下载
下载地址为:https://www.scala-lang.org/download/2.12.8.html
集体为 Win 11,间接下载 Windows 装置版本:
集体是 window 零碎,间接点击下一步即可,装置过程这里省略了。
配置 SCALA_HOME
装置实现之后是在对应的操作系统配置环境变量,集体在 Path 变量中减少SCALA_HOME,并且指定地址即可。因为集体是 Windows 装置版本装置,曾经主动配置了环境变量。装置实现之后后果如下:
C:\Users\adong>scala -version
Scala code runner version 2.12.8 -- Copyright 2002-2018, LAMP/EPFL and Lightbend, Inc.
装置 gradle 6.6
下载地址:https://services.gradle.org/distributions/
咱们抉择想要装置的公布版本,gradle-x.x-bin.zip 是须要下载的装置公布版,gradle-x.x-src.zip 是源码,gradle-x.x-all.zip 则是下载全副的文件。这里我抉择的是gradle-6.6.1-bin.zip
。
配置 Gradle 环境变量
间接把 gradle 的装置目录 bin 地址贴到环境变量的 Path 当中。
E:\adongstack\mysoft\gradle-6.6.1\bin
而后是在 cmd
当中验证即可:
C:\Users\adong>gradle -v
Welcome to Gradle 6.6.1!
Here are the highlights of this release:
- Experimental build configuration caching
- Built-in conventions for handling credentials
- Java compilation supports --release flag
For more details see https://docs.gradle.org/6.6.1/release-notes.html
------------------------------------------------------------
Gradle 6.6.1
------------------------------------------------------------
Build time: 2020-08-25 16:29:12 UTC
Revision: f2d1fb54a951d8b11d25748e4711bec8d128d7e3
Kotlin: 1.3.72
Groovy: 2.5.12
Ant: Apache Ant(TM) version 1.10.8 compiled on May 10 2020
JVM: 1.8.0_333 (Oracle Corporation 25.333-b02)
OS: Windows 11 10.0 amd64
装置 Zookeeper 3.4.14
集体倡议装置到虚拟机的 Linux 零碎,装置的操作如下:
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz
国内的访问速度比较慢,这里间接上外网拖了一个安装包放到网盘,读者依据网盘下载即可:
链接:https://pan.baidu.com/s/1R6zz5pffuwV8D7B-N6PV7Q?pwd=wopn
提取码:wopn
创立 data 文件夹
在 Zookeeper 3.4.14 的目录当中筹备 data
文件夹。
[zxd@localhost zookeeper-3.4.14]$ mkdir data
mkdir: cannot create directory‘data’: Permission denied
[zxd@localhost zookeeper-3.4.14]$ sudo mkdir data
[sudo] password for zxd:
[zxd@localhost zookeeper-3.4.14]$ ls
bin conf data dist-maven
批改 zoo.cfg 中的 data 属性
咱们将 Zookeeper 的数据目录执行刚刚新建的 data 文件夹。
cd conf
vim zoo_sample.cfg zoo.cfg
vim zoo_sample.cfg zoo.cfg
,咱们批改相干的配置,默认目录在/tmp/zookeeper
,这里集体改为/opt/zookeeper/zookeeper-3.4.14/data
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
#dataDir=/tmp/zookeeper
dataDir=/opt/zookeeper/zookeeper-3.4.14/data
# the port at which the clients will connect
clientPort=2181
批改之后咱们应用-x
利用批改。
启动 Zookeeper
编辑实现之后,进入到装置目录的 bin 目录当中启动 Zookeeper。
[zxd@localhost zookeeper-3.4.14]$ cd bin/
[zxd@localhost bin]$ sudo ./zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.4.14/bin/../conf/zoo.cfg
grep: /opt/zookeeper/zookeeper-3.4.14/bin/../conf/zoo.cfg: No such file or directory
mkdir: cannot create directory‘’: No such file or directory
Starting zookeeper ... STARTED
发现没有启动胜利,依据提醒咱们须要 zoo.cfg 的配置,这部分为 Zk 的内容,这里就间接跳过了。
Kafka 2.7.2 源码下载
下载 kafka-2.7.2-src.tgz
包之后,解压到对应的目录,接下来须要导入到 Idea 当中。
导入 Idea
下载源码之后在 Idea 当中导入 Kafka 我的项目。
导入我的项目之后,咱们进入到 Setting,设置 Gradle 的 Home 地址。
这里把 Gradle 的 Home 地址设置为本人本机装置的,不应用 Iead 自带的。
批改 build.gradle
接下来还不能导 jar 包,须要把镜像文件下载服务器更换为国内的私服,否则会相当慢,间接导致 “time out” 报错。进入 kafka 源码包,批改 build.gradle 文件,在原来配置上,增加阿里的私服配置。
上面的内容复制到 build.gradle 文件的对应地位:
maven {url 'http://maven.aliyun.com/nexus/content/groups/public/'}
maven {url 'http://maven.aliyun.com/nexus/content/repositories/jcenter'}
文件当中查找
buildscript
能够疾速定位。
留神最初须要 调整一下地位,优先从阿里云的库获取:
buildscript {
repositories {
maven {url 'http://maven.aliyun.com/nexus/content/groups/public/'}
maven {url 'http://maven.aliyun.com/nexus/content/repositories/jcenter'}
mavenCentral()
jcenter()
maven {url "https://plugins.gradle.org/m2/"}
}
......
接着还须要复制到 allproject
当中:
maven {url 'http://maven.aliyun.com/nexus/content/groups/public/'}
maven {url 'http://maven.aliyun.com/nexus/content/repositories/jcenter'}
复制到的地位大略如下:
allprojects {
repositories {
maven {url 'http://maven.aliyun.com/nexus/content/groups/public/'}
maven {url 'http://maven.aliyun.com/nexus/content/repositories/jcenter'}
mavenCentral()}
替换之后急躁期待依赖下载实现:
在依赖装置实现之后,咱们能够点一下 Idea 顶部的小锤子,最初的提醒编译胜利:
装置 Idea scala 插件
Idea 导入源码之后的 .scala
文件是没有语法提醒的,咱们装置插件之后能够失常浏览 scala 的源码:
插件装置实现之后须要重启。拜访kafka.scala
,如果相干的代码呈现色彩和语法提醒阐明插件装置胜利:
抉择 scala jdk
如果读者是第一次搭建 Kafka 和应用 scala,大概率会呈现上面的提醒:
依据提醒设置 scala jdk
的装置目录:
C:\Program Files (x86)\scala\bin
设置实现之后所有就功败垂成了。
编译和构建 Kafka 源码
这里集体不太熟悉 Idea 对于 gradle 的应用,最初用了 gradle 的命令进行构建,构建之后呈现上面的内容示意正确:
gradle
命令是用来下载和更新 Gradle 套件(Gradle Wrapper)的,gradle jar
是用 Gradle 套件构建 Kafka 工程,生成 Jar 文件。
上面是应用gradle
命令的后果:
adong@Adong-pc MINGW64 /e/adongstack/project/selfUp/kafka-2.7/kafka-2.7.2-src
$ gradle
Starting a Gradle Daemon, 1 busy Daemon could not be reused, use --status for details
> Configure project :
Building project 'core' with Scala version 2.13.3
Building project 'streams-scala' with Scala version 2.13.3
> Task :help
Welcome to Gradle 6.6.1.
To run a build, run gradle <task> ...
To see a list of available tasks, run gradle tasks
To see a list of command-line options, run gradle --help
To see more detail about a task, run gradle help --task <task>
For troubleshooting, visit https://help.gradle.org
Deprecated Gradle features were used in this build, making it incompatible with Gradle 7.0.
Use '--warning-mode all' to show the individual deprecation warnings.
See https://docs.gradle.org/6.6.1/userguide/command_line_interface.html#sec:command_line_warnings
BUILD SUCCESSFUL in 44s
1 actionable task: 1 executed
上面应用 gradle jar
的编译后果:
adong@Adong-pc MINGW64 /e/adongstack/project/selfUp/kafka-2.7/kafka-2.7.2-src
$ gradle jar
> Configure project :
Building project 'core' with Scala version 2.13.3
Building project 'streams-scala' with Scala version 2.13.3
> Task :clients:processMessages
MessageGenerator: processed 121 Kafka message JSON files(s).
> Task :raft:processMessages
MessageGenerator: processed 1 Kafka message JSON files(s).
<=------------> 12% EXECUTING [2m 6s]
> Task :clients:processTestMessages
MessageGenerator: processed 2 Kafka message JSON files(s).
<==-----------> 21% EXECUTING [3m 10s]
> Task :streams:processMessages
MessageGenerator: processed 1 Kafka message JSON files(s).
Deprecated Gradle features were used in this build, making it incompatible with Gradle 7.0.
Use '--warning-mode all' to show the individual deprecation warnings.
See https://docs.gradle.org/6.6.1/userguide/command_line_interface.html#sec:command_line_warnings
BUILD SUCCESSFUL in 3m 58s
87 actionable tasks: 87 executed
最初应用上面的命令:
gradle clean build -x test
然而这种命令构建过程会呈现一些报错,能够把报错所须要的依赖 jar 包放到上面:
E:\adongstack\project\selfUp\kafka-2.7\kafka-2.7.2-src\gradle\wrapper
要构建整个 Kafka 工程并打包出一个可运行的二进制环境命令如下
gradle clean releaseTarGz
其余构建命令
还有其余 gradle 的构建命令:
构建 jar 包并运行
./gradle jar
构建我的项目,看你是 idea 工具还是 eclipse
./gradle idea
./gradle eclipse
构建源码包
./gradle srcJar
构建 javadoc 文档
./gradle aggregatedJavadoc
清理并构建
./gradle clean
单元测试
Kafka 当中同样存在很多单元测试,上面是一些外围模块的单元测试命令。
core 模块单元测试:
gradle core:test
client 模块单元测试:
gradle clients:test
模块当中的某个子模块单元测试:
gradle connect:[submodule]:test
Connect 工程下细分了多个子模块,比方 api、runtime 等,须要显式地指定要 测试的子模块名能力进行测试。
stream 模块单元测试。
gradle connect:[submodule]:test
集体理论应用之后会发现有局部报错信息:
除了下面的整个模块的单元测试,如果只想要测试某个模块下的某个类,能够应用上面的办法:
独自对某一个具体的测试用例进行测试,比方测试 LogTest 类:
gradle core:test --tests kafka.log.LogTest
集体的执行后果如下:
$ gradle core:test --tests kafka.log.LogTest
> Configure project :
Building project 'core' with Scala version 2.13.3
Building project 'streams-scala' with Scala version 2.13.3
> Task :core:test
kafka.log.LogTest.testLogDelete failed, log available in E:\adongstack\project\selfUp\kafka-2.7\kafka-2.7.2-src\core\build\reports\testOutput\kafka.log.LogTest.testLogDelete.test.stdout
kafka.log.LogTest > testLogDelete FAILED
org.apache.kafka.common.errors.KafkaStorageException: Error while deleting log for kafka-936488 in dir C:\Users\adong\AppData\Local\Temp\kafka-7162272829353105740
Caused by:
java.nio.file.FileSystemException: C:\Users\adong\AppData\Local\Temp\kafka-7162272829353105740\kafka-936488\00000000000000000000.timeindex -> C:\Users\adong\AppData\Local\Temp\kafka-7162272829353105740\kafka-936488\00000000000000000000.timeindex.deleted: ▒▒һ▒▒▒▒▒▒▒▒▒▒ʹ▒ô▒▒ļ▒▒▒▒▒▒▒▒▒▒▒▒ʡ▒
Kafka 目录构造
bin 目录:保留 Kafka 工具行脚本,咱们熟知的 kafka-server-start 和 kafka-consoleproducer 等脚本都寄存在这。
clients 目录:保留 Kafka 客户端代码,比方生产者和消费者的代码都在该目录下。
config 目录:保留 Kafka 的配置文件,其中比拟重要的配置文件是 server.properties。
connect 目录 :保留 Connect 组件的源代码。Kafka Connect 组件是用来实现 Kafka 与内部零碎之间的 实时数据传输 的。
core 目录:保留 Broker 端代码。Kafka 服务器端代码全副保留在该目录下。
streams 目录 :保留 Streams 组件的源代码。Kafka Streams 是实现 Kafka 实时流解决 的组件。
server 目录:顾名思义,它是 Kafka 的服务器端主代码,外面的类十分多,很多要害的 Kafka 组件都寄存在这里,比方状态机、Purgatory 延时机制等。
Kafka Streams is a client library for building applications and microservices, where the input and output data are stored in Kafka clusters.
提供一个基于 Kafka 的流式解决类库,间接提供具体的类给开发者调用,整个利用的运行形式次要由开发者管制,方便使用和调试。
Kafka Streams 是一个用来构建流处理程序的库,特地是其输出是一个 Kafka topic,输入是另一个 Kafka topic 的程序(或者是调用内部服务,或者是更新数据库,或者其它)。它使得你以一种分布式以及容错的形式来做这件事件。
coordinator 包:保留了消费者端的 GroupCoordinator 代码和用于事务的 TransactionCoordinator 代码。对 coordinator 包进行剖析,特地是对消费者端的 GroupCoordinator 代码进行剖析,是咱们弄明确 Broker 端协调者组件设计原理的关 键。
其余目录构造
除了下面的外围目录之外,咱们还能够在我的项目根门路看到一些其余模块:
network 包:封装了 Kafka 服务器端网络层的代码,特地是 SocketServer.scala 这个文件,是 Kafka 实现 Reactor 模式的具体操作类,十分值得一读。
consumer 包:前面会抛弃该包,用 clients 包下 consumer 相干类代替。
tools 包:工具类。
log 包:保留了 Kafka 最外围的日志构造代码,包含日志、日志段、索引文件等, 另外,该包下还封装了 Log Compaction 的实现机制,是十分重要的源码包。
checkstyle 目录:代码标准,自动化检测。
Checkstyle 是什么,相似于代码标准的自动化检测插件,国内最为经典的是阿里巴巴的规约插件,而最闻名的查看插件为
google style guide
,Checkstyle 就是以相似这种格调开发出的一个自动化插件,来辅助判断代码格局是否满足标准。
该目录下的文件定义了工程代码格局的标准,咱们能够在 build.gradle 中看到相干 checkstyle 的配置和自动化代码格式化配置:
checkstyle 配置(build.gradle)
def checkstyleConfigProperties(configFileName) {
[importControlFile: "$rootDir/checkstyle/$configFileName",
suppressionsFile: "$rootDir/checkstyle/suppressions.xml",
headerFile: "$rootDir/checkstyle/java.header"]
}
checkstyle {configProperties = checkstyleConfigProperties("import-control-core.xml")
}
connect 目录:保留 Connect 组件的源代码。Kafka Connect 组件是用来实现 Kafka 与内部零碎之间的实时数据传输的。
docs 目录:Kafka 设计文档以及组件相干结构图。
examples 目录:Kafka 样例相干目录。
gradle 目录:gradle 的脚本和依赖包定义等相干文件。
jmh-benchmarks 目录:Kafka 代码微基准测试相干类。
JMH,即 Java Microbenchmark Harness,是 专门用于代码微基准测试的工具套件 。Micro Benchmark 是什么意思?简略的来说就是 基于办法层面的基准测试,精度能够达到微秒级。当你定位到热点办法,心愿进一步优化办法性能的时候,就能够应用 JMH 对优化的后果进行量化的剖析。
JMH 比拟典型的利用场景有:
- 精确的晓得某个办法须要执行多长时间,以及执行工夫和输出之间的相关性;
- 比照接口不同实现在给定条件下的吞吐量,找到最优实现。
kafka-logs 目录:server.properties 文件中配置 log.dirs 生成的目录。
log4j-appender 目录:
A log4j appender that produces log messages to Kafka
这个目录外面就一个 KafkaLog4jAppender 类。
raft 目录:raft 一致性协定相干模块。
tests 目录:此目录的内容介绍如何进行 Kafka 系统集成和性能测试。
tools 目录:工具类模块。
vagrant 目录:介绍如何在 Vagrant 虚构环境中运行 Kafka,提供了相干的脚本文件和阐明文档。
Vagrant 是一个基于 Ruby 的工具,用于创立和部署虚拟化开发环境。它应用 Oracle 的开源 VirtualBox 虚拟化零碎,应用 Chef 创立自动化虚拟环境。
参考
https://segmentfault.com/a/1190000040790524