Kafka 集群参数

重要的配置 : Broker 端参数，主题级别的参数、JVM 端参数、操作系统级别的参数

Broker 端参数

存储配置

• log.dirs：指定 Broker 要用的若干个文件目录路径 (自指定)
• log.dir：只能指定单个路径
• 设置了 log.dirs ，就不用设置 log.dir

线上环境要 log.dirs 配置多个路径

• 如 ：/home/kafka1,/home/kafka2,/home/kafka3

目录挂载在不同的物理磁盘的好处：

• 提升读写性能：多块物理磁盘同时读写
• 实现故障转移 (Failover) : Kafka 1.1 引入。坏磁盘的数据会自动转到其他正常的磁盘上，并正常工作

ZooKeeper 配置

ZooKeeper : 分布式协调框架，负责协调管理并保存 Kafka 集群的所有元数据信息

• 元数据信息 : Broker 运行、Topic，每个 Topic 的分区，分区的 Leader 副本

Kafka/ZK 连接 ：zookeeper.connect

• 如: zk1:2181,zk2:2181,zk3:2181

多 Kafka集群用单 ZK 集群 :

• 用 chroot 起别名
• zk1:2181,zk2:2181,zk3:2181/kafka1
• zk1:2181,zk2:2181,zk3:2181/kafka2

Broker 连接配置

Broker 连接 ：客户端或其他 Broker 通信

• listeners：监听器，外部连接者通过什么协议访问指定主机名和端口开放的 Kafka 服务
• advertised.listeners：比 listeners 多 advertised。该组监听器对外网公开
• host.name/port：(忘掉，过时参数)

监听器的构成 :

• 若干个逗号分隔的三元组，每个三元组的格式为 <协议名称, 主机名, 端口号>
• 协议名称，如 : PLAINTEXT : 明文传输 ，SSL : SSL 或 TLS 加密传输
• 自定义协议，如 : CONTROLLER: //localhost:9092

自定义协议名称

• listener.security.protocol.map: 指定安全协议
• 如 : listener.security.protocol.map=CONTROLLER:PLAINTEXT : CONTROLLER 自定义协议，用明文不加密传输数据

Topic 管理配置

• auto.create.topics.enable：是否允许自动创建 Topic (建议 : false : 不允许自动创建 Topic)
• unclean.leader.election.enable：是否允许 Unclean Leader 选举 ( 建议 :false : 保证数据不易丢失)
• auto.leader.rebalance.enable ：是否定期 Leader 选举 (建议 :false : 避免 Leader 重选举)

数据留存配置

• log.retention.{hour|minutes|ms}：控制消息保存多长。优先级 : ms > minutes > hour ( 默认 :log.retention.hour=168(7 天) )
• log.retention.bytes：指定 Broker 总磁盘容量大小 ( 默认: -1 : 不限制 ; 构建多租户时具体指定 )
• message.max.bytes：控制 Broker 能接收的最大消息大小 (默认 : 1000012 , 建议设大)

Topic 级别参数

当同时设置 Topic 级别参数和全局 Broker 参数时 :

• Topic 级别参数优先级 > 全局 Broker 参数

Topic 参数 :

• retention.ms：该 Topic 消息保存时长 (默认: 7 天)
• retention.bytes：该 Topic 预留多大的磁盘空间 (默认: -1 : 无限用磁盘空间)
• max.message.bytes : 接收该 Topic 的最大消息大小

Topic 级别参数设置 :

• 创建 Topic 时设置
• 修改 Topic 时设置

kafka-topics 创建 Topic : 设置半年数据，最大消息 5MB :

• 15552000000 = 180 * 24 * 60 * 60 * 1000
• 5242880 = 5 * 1024 * 1024

bin/kafka-topics.sh \
--bootstrap-server localhost:9092 \
--create--topic transaction \
--partitions 1 \
--replication-factor 1 \
--config retention.ms=15552000000 \
--config max.message.bytes=5242880

kafka-configs 修改 Topic : 设置最大消息 10MB

• 10485760 = 10 * 1024 * 1024

bin/kafka-configs.sh \
--zookeeper localhost:2181 \
--entity-type topics \
--entity-name transaction \
--alter --add-config max.message.bytes=10485760

JVM 参数

Kafka 服务器端用 Scala 写 , 依然运行在 JVM 上 , 使用 Java 8 上

• JVM 堆大小设置成 6GB
• 用 G1 收集器

export KAFKA_HEAP_OPTS=--Xms6g --Xmx6gexport KAFKA_JVM_PERFORMANCE_OPTS= -server \
-XX:+UseG1GC -XX:MaxGCPauseMillis=20 \
-XX:InitiatingHeapOccupancyPercent=35 \
-XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=truebin/kafka-server-start.sh config/server.properties

• MaxGCPauseMillis : 每次GC最大的停顿毫秒
• InitiatingHeapOccupancyPercent : 8b12前 : 整堆使用量与堆总容量的比值 ; 8b12 后 : 老年代大小与堆总容量的比值
• ExplicitGCInvokesConcurrent : (不懂 , 问题点) , 网上说 cms 下才生效 , g1 什么关系
• java.awt.headless=true : 强制用 Headless 的 AWT 实现类，避免缺失图形环境而出错

操作系统参数

OS 参数 :

• 文件描述符限制：ulimit -n 1000000 , 调大避免 Too many open files
• 文件系统类型 : 生产环境用高效 XFS
• Swappiness : vm.swappiness = 1 , 避免性能下降
• Flush 落盘时间 : 调大提交间隔 (默认 : 5s) 换性能