消息中央件-Kafka2-3.9.0源码构建

消息中央件-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 ->roject 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文件参数阐明

1. # 每一个Broker在集群中的唯一标识。即使Broker的IP地址发生了变化，broker.id只要没变，则不会影响consumers的消息情况
2. broker.id=0
4. # 是否允许Topic被删除。如果是false，使用管理员工具删除Topic的时候，Kafka并不会处理此操作
5. # delete.topic.enable=true
7. # Kafka服务端是否可以根据请求自动创建Topic，默认是true。如果打开此选项，下面三种请求会触发Topic自动创建：
8. # ①Producer向某个不存在的Topic写入消息
9. # ②Consumer从某个不存在的Topic读取消息
10. # ③Consumer从某个不存在的Topic读取消息
11. # 建议将此选项设置为false，并在使用Topic之前手动创建
12. # auto.create.topics.enable=true
14. # Kafka Server使用的协议、主机名以及端口的格式如下：
15. # listeners = security_protocol://host_name:port
16. # 参考示例:
17. # listeners = PLAINTEXT://your.host.name:9092
18. #listeners=PLAINTEXT://:9092 #这是默认配置，使用PLAINTEXT，端口是9092
20. # 接收请求的线程数
21. num.network.threads=3
23. # 执行请求的线程数
24. num.io.threads=8
26. # 每个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，就使用操作系统的默认值
27. socket.send.buffer.bytes=102400
29. # TCP连接的SO_RCVBUF缓冲区大小，默认102400，单位是字节如果是-1，就使用操作系统的默认值
30. socket.receive.buffer.bytes=102400
32. # 请求的最大长度
33. socket.request.max.bytes=104857600
35. # 用于存储log文件的目录，可以将多个目录通过逗号分隔，形成一个目录列表
36. log.dirs=/tmp/kafka-logs
38. # 每个Topic默认的partition数量，默认值是1
39. num.partitions=1
41. # 用来恢复log文件以及关闭时将log数据刷新到磁盘的线程数量，每个目录对应num.recovery.threads.per.data.dir个线程
42. num.recovery.threads.per.data.dir=1
44. # 每隔多少个消息触发一次flush操作，将内存中的消息刷新到硬盘上
45. #log.flush.interval.messages=10000
47. # 每隔多少毫秒触发一次flush操作，将内存中的消息刷新到硬盘上
48. #log.flush.interval.ms=1000
50. # 上面这两个配置是全局的，可以在Topic中重新设置，并覆盖这两个配置
51. # 下面有两种配置，一种是基于时间的策略，另一种是基于日志文件大小的策略，两种策略同是配置的话，只要满足其中一种策略，则触发Log删除的操作。删除操作总是先删除最旧的日志
53. # 消息在Kafka中保存的时间，168小时之前的log，可以被删除掉
54. log.retention.hours=168
56. # 当剩余空间低于log.retention.bytes字节，则开始删除log
57. #log.retention.bytes=1073741824
59. # segment日志文件大小的上限值。当超过这个值时，会创建新的segment日志文件
60. log.segment.bytes=1073741824
62. # 每隔300000ms，logcleaner线程将检查一次，看是否符合上述保留策略的消息可以被删除
63. log.retention.check.interval.ms=300000
65. # Kafka依赖的ZooKeeper集群地址，可以配置多个ZooKeeper地址，使用逗号隔开
66. zookeeper.connect=localhost:2181
68. # ZooKeeper连接的超时时间
69. zookeeper.connection.timeout.ms=6000

复制代码

设置Kafka的启动参数,Kafka服务端的入口类是kafka.Kafka.scala

除了指定入口类，还必要指定server.properties设置文件所在的位置，即在Kafka源代码的config目次下,并设置运行设置

由于kafka依赖Zookeeper, 因此必要先将本地的zookeeper启动起来,默认端口2181

末了启动Kafka，启动完成

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。