消息中间件-Kafka2-3.9.0源码构建
1、软件环境
- JDK Version 1.8
- Scala Version 2.12.0
- Kafka-3.9.0 源码包
下载地址:https://downloads.apache.org/kafka/3.9.0/kafka-3.9.0-src.tgz - Gradle Version >= 8.8
- Apache Zookeeper 3.7.0
2、源码编译
- 打开源码根目录
- 修改build.gradle 文件, 添加阿里云仓库
- 在源码根目录下执行编译命令
编译IDEA项目执行: gradle idea
编译Eclipse项目执行: gradle eclipse
3、源码构建成功
- 源码构建时间, 需要耐心等待,大概10分钟左右
- 构建过程中出现无法下载包的情况,可能是网络较慢, 多次执行步骤2-3即可
- 源码构建成功标志
4、源码导入
将源码导入IDEA或Eclipse中, IDEA或Eclipse需要安装scala插件,否则项目无法识别。不要使用File->Open 要选择 File ->Project from Existing sources , 如果没有这个选项 则选择New -> Project from Existing sources -> import gradle 将项目导入。 导入后打开Setting配置,修改gradle配置,如下图
修改scala配置, 如下图
点击ok即可, 等待项目编译。
导入成功.
5、配置、启动Kafka
在Kafka服务端使用log4j输出日志,启动前需要在core/src/main下创建resources目录,将config目录下的log4j.properties 文件复制到此文件夹中,并在build.gradle文件中添加log4j的依赖,完成上述操作运行,才不会出现日志无法加载的错误。
server.properties是Kafka的主要配置文件,下面简单介绍其中的相关配置项的含义。必须修改的配置项就是log.dirs,其他配置可以根据需求自行修改。在根目录下创建logs目录,让后将配置文件中的日志目录指向该目录。
server.properties文件参数说明
# 每一个Broker在集群中的唯一标识。即使Broker的IP地址发生了变化,broker.id只要没变,则不会影响consumers的消息情况
broker.id=0# 是否允许Topic被删除。如果是false,使用管理员工具删除Topic的时候,Kafka并不会处理此操作
# delete.topic.enable=true# Kafka服务端是否可以根据请求自动创建Topic,默认是true。如果打开此选项,下面三种请求会触发Topic自动创建:
# ①Producer向某个不存在的Topic写入消息
# ②Consumer从某个不存在的Topic读取消息
# ③Consumer从某个不存在的Topic读取消息
# 建议将此选项设置为false,并在使用Topic之前手动创建
# auto.create.topics.enable=true# Kafka Server使用的协议、主机名以及端口的格式如下:
# listeners = security_protocol://host_name:port
# 参考示例:
# listeners = PLAINTEXT://your.host.name:9092
#listeners=PLAINTEXT://:9092 #这是默认配置,使用PLAINTEXT,端口是9092# 接收请求的线程数
num.network.threads=3# 执行请求的线程数
num.io.threads=8# 每个TCP socket在内核中都有一个发送缓冲区(SO_SNDBUF)和一个接收缓冲区(SO_RCVBUF)。接收缓冲区把数据缓存入内核,应用进程一直没有调用read进行读取的话,此数据会一直缓存在相应socket的接收缓冲区内。再啰嗦一点,不管进程是否读取socket,对端发来的数据都会经由内核接收并且缓存到socket的内核接收缓冲区之中。read所做的工作,就是把内核缓冲区中的数据复制到应用层用户的buffer里面,仅此而已。进程调用send发送的数据的时候,一般情况下,将数据复制进入socket的内核发送缓冲区之中,然后send便会在上层返回。换句话说,send返回之时,数据不一定会发送到对端去,send仅仅是把应用层buffer的数据复制进socket的内核发送buffer中TCP连接的SO_SNDBUF缓冲区大小,默认102400,单位是字节如果是-1,就使用操作系统的默认值
socket.send.buffer.bytes=102400# TCP连接的SO_RCVBUF缓冲区大小,默认102400,单位是字节如果是-1,就使用操作系统的默认值
socket.receive.buffer.bytes=102400# 请求的最大长度
socket.request.max.bytes=104857600# 用于存储log文件的目录,可以将多个目录通过逗号分隔,形成一个目录列表
log.dirs=/tmp/kafka-logs# 每个Topic默认的partition数量,默认值是1
num.partitions=1# 用来恢复log文件以及关闭时将log数据刷新到磁盘的线程数量,每个目录对应num.recovery.threads.per.data.dir个线程
num.recovery.threads.per.data.dir=1# 每隔多少个消息触发一次flush操作,将内存中的消息刷新到硬盘上
#log.flush.interval.messages=10000# 每隔多少毫秒触发一次flush操作,将内存中的消息刷新到硬盘上
#log.flush.interval.ms=1000# 上面这两个配置是全局的,可以在Topic中重新设置,并覆盖这两个配置
# 下面有两种配置,一种是基于时间的策略,另一种是基于日志文件大小的策略,两种策略同是配置的话,只要满足其中一种策略,则触发Log删除的操作。删除操作总是先删除最旧的日志# 消息在Kafka中保存的时间,168小时之前的log,可以被删除掉
log.retention.hours=168# 当剩余空间低于log.retention.bytes字节,则开始删除log
#log.retention.bytes=1073741824# segment日志文件大小的上限值。当超过这个值时,会创建新的segment日志文件
log.segment.bytes=1073741824# 每隔300000ms,logcleaner线程将检查一次,看是否符合上述保留策略的消息可以被删除
log.retention.check.interval.ms=300000# Kafka依赖的ZooKeeper集群地址,可以配置多个ZooKeeper地址,使用逗号隔开
zookeeper.connect=localhost:2181# ZooKeeper连接的超时时间
zookeeper.connection.timeout.ms=6000
设置Kafka的启动参数,Kafka服务端的入口类是kafka.Kafka.scala
除了指定入口类,还需要指定server.properties配置文件所在的位置,即在Kafka源代码的config目录下,并设置运行配置
由于kafka依赖Zookeeper, 因此需要先将本地的zookeeper启动起来,默认端口2181
最后启动Kafka,启动完成