Broker端参数

也称 静态参数（Static Configs)
介绍的所有参数都是那些要修改默认值的参数;因为它们默认值不适合一般的生产环境。

• 静态参数
  • 必须在 Kafka 的配置文件 server.properties 中进行设置的参数，不管你是新增、修改还是删除。
  • 同时，你必须重启 Broker 进程才能令它们生效。

与存储信息相关的参数

• **log.dirs**：非常重要的参数，指定了 Broker 需要使用的若干个文件目录路径。这个参数是没有默认值的，必须亲自指定。
• **log.dir**：说明它只能表示单个路径，它是补充上一个参数用的。
• 怎么设置呢？
  • 只要设置log.dirs，即第一个参数就好了，不要设置log.dir
  • 更重要的是，在线上生产环境中一定要为log.dirs配置多个路径，具体格式是一个 CSV 格式，也就是用逗号分隔的多个路径
  • 有条件的话你最好保证这些目录挂载到不同的物理磁盘上
• 两个好处
  • 提升读写性能：比起单块磁盘，多块物理磁盘同时读写数据有更高的吞吐量。
  • 能够实现故障转移：即 Failover。

    这是 Kafka 1.1 版本新引入的强大功能。要知道在以前，只要 Kafka Broker 使用的任何一块磁盘挂掉了，整个 Broker 进程都会关闭。但是自 1.1 开始，这种情况被修正了，坏掉的磁盘上的数据会自动地转移到其他正常的磁盘上，而且 Broker 还能正常工作。还记得上一期我们关于 Kafka 是否需要使用 RAID 的讨论吗？这个改进正是我们舍弃 RAID 方案的基础：没有这种 Failover 的话，我们只能依靠 RAID 来提供保障。

与 ZooKeeper 相关的参数

• ZooKeeper 是做什么的呢

  • 它是一个分布式协调框架，负责协调管理并保存 Kafka 集群的所有元数据信息
  • 比如集群都有哪些 Broker 在运行、创建了哪些 Topic，每个 Topic 都有多少分区以及这些分区的 Leader 副本都在哪些机器上等信息。

• zookeeper.connect: 是一个 CSV 格式的参数；2181 是 ZooKeeper 的默认端口。

• 让多个 Kafka 集群使用同一套 ZooKeeper 集群，那么这个参数应该怎么设置呢？

  • chroot 就派上用场了。这个 chroot 是 ZooKeeper 的概念，类似于别名。

• 同一zookeeper配置多套kafka

  • 有两套 Kafka 集群，假设分别叫它们 kafka1 和 kafka2
  • 两套集群的zookeeper.connect参数可以这样指定：
  • zk1:2181,zk2:2181,zk3:2181/kafka1和zk1:2181,zk2:2181,zk3:2181/kafka2

与 Broker 连接相关的参数

• 即客户端程序或其他 Broker 如何与该 Broker 进行通信的设置

• listeners: 监听器，其实就是告诉外部连接者要通过什么协议访问指定主机名和端口开放的 Kafka 服务。

• advertised.listeners: 这组监听器是 Broker 用于对外发布的。

• host.name/port: 过期的参数

• 监听器的概念

  • 从构成上来说，它是若干个逗号分隔的三元组，每个三元组的格式为<协议名称，主机名，端口号>。
  • 这里的协议名称可能是标准的名字，比如 PLAINTEXT 表示明文传输、SSL 表示使用 SSL 或 TLS 加密传输等；也可能是你自己定义的协议名字，比如CONTROLLER: //localhost:9092。
  • 必须还要指定listener.security.protocol.map参数告诉这个协议底层使用了哪种安全协议;比如指定listener.security.protocol.map=CONTROLLER:PLAINTEXT表示CONTROLLER这个自定义协议底层使用明文不加密传输数据。
  • 主机名这个设置中我到底使用 IP 地址还是主机名。这里我给出统一的建议：最好全部使用主机名，即 Broker 端和 Client 端应用配置中全部填写主机名。Broker 源代码中也使用的是主机名，如果你在某些地方使用了 IP 地址进行连接，可能会发生无法连接的问题。

与 Topic 管理相关的参数

• auto.create.topics.enable:是否允许自动创建 Topic。
  • 最好设置成 false，即不允许自动创建 Topic。
  • 对于那些大公司而言，每个部门被分配的 Topic 应该由运维严格把控，决不能允许自行创建任何 Topic。
• unclean.leader.election.enable:是否允许 Unclean Leader 选举。
  • 关闭 Unclean Leader 选举的
  • 只有保存数据比较多的那些副本才有资格竞选，那些落后进度太多的副本没资格做这件事。
  • 假设那些保存数据比较多的副本都挂了怎么办？我们还要不要进行 Leader 选举了？
  • 如果设置成 false，那么就坚持之前的原则，坚决不能让那些落后太多的副本竞选 Leader。这样做的后果是这个分区就不可用了，因为没有 Leader 了。
  • 如果是 true，那么 Kafka 允许你从那些“跑得慢”的副本中选一个出来当 Leader。这样做的后果是数据有可能就丢失了，因为这些副本保存的数据本来就不全
  • 在最新版的 Kafka 中默认就是 false; 建议还是显式地把它设置成 false 吧。
• auto.leader.rebalance.enable:是否允许定期进行 Leader 选举。
  • 设置它的值为 true 表示允许 Kafka 定期地对一些 Topic 分区进行 Leader 重选举;
  • 严格来说它与上一个参数中 Leader 选举的最大不同在于，它不是选 Leader，而是换 Leader！
  • 比如 Leader A 一直表现得很好，但若auto.leader.rebalance.enable=true，那么有可能一段时间后 Leader A 就要被强行卸任换成 Leader B。
  • 建议你在生产环境中把这个参数设置成 false。

与数据留存相关的参数

• **log.retention.{hours|minutes|ms}**：这是个“三兄弟”，都是控制一条消息数据被保存多长时间。从优先级上来说 ms 设置最高、minutes 次之、hours 最低。
  • 这里设置的是集群级别的，Kafka 还支持Topic级别的保存时间设置。
  • 比如log.retention.hours=168表示默认保存 7 天的数据，自动删除 7 天前的数据。
• **log.retention.bytes**：指定 Broker 为消息保存的总磁盘容量大小。
  • 这个值默认是 -1，表明你想在这台 Broker 上保存多少数据都可以
  • 这个参数真正发挥作用的场景其实是在云上构建多租户的 Kafka 集群
  • 设想你要做一个云上的 Kafka 服务，每个租户只能使用 100GB 的磁盘空间，为了避免有个“恶意”租户使用过多的磁盘空间，设置这个参数就显得至关重要了。
• **message.max.bytes**：控制 Broker 能够接收的最大消息大小；针对单数据较大的业务需要调整该大小
  • 不能使用默认值的参数，这个参数也是一样，默认的 1000012 太少了，还不到 1MB。
  • 在线上环境中设置一个比较大的值还是比较保险的做法。毕竟它只是一个标尺而已，仅仅衡量 Broker 能够处理的最大消息大小，即使设置大一点也不会耗费什么磁盘空间的。

Topic级别参数

Kafka 提供了专门的 kafka-configs 命令来修改它们

• 如果同时设置了 Topic 级别参数和全局 Broker 参数，到底听谁的呢？哪个说了算呢？
  • Topic 级别参数会覆盖全局 Broker 参数的值，而每个 Topic 都能设置自己的参数值，这就是所谓的 Topic 级别参数。
  • topic参数高于broker参数

    与保存消息相关的参数

• retention.ms: 规定了该 Topic 消息被保存的时长。
  • 默认是 7 天，即该 Topic 只保存最近 7 天的消息。一旦设置了这个值，它会覆盖掉 Broker 端的全局参数值。
• retention.bytes: 规定了要为该 Topic 预留多大的磁盘空间。
  • 和全局参数作用相似，这个值通常在多租户的 Kafka 集群中会有用武之地。当前默认值是 -1，表示可以无限使用磁盘空间。

与能处理的消息大小相关参数

• **max.message.bytes**：决定了 Kafka Broker 能够正常接收该 Topic 的最大消息大小。
  • 该参数跟 message.max.bytes 参数的作用是一样的，只不过 max.message.bytes 是作用于某个 topic，而 message.max.bytes 是作用于全局。

怎么设置 Topic 级别参数

• 有两种方式可以设置：
  • 创建 Topic 时进行设置
  • 修改 Topic 时设置
• 如何在创建 Topic 时设置这些参数

1

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 级别参数

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

  最好始终坚持使用第二种方式来设置，并且在未来，Kafka 社区很有可能统一使用kafka-configs脚本来调整 Topic 级别参数。

JVM 端参数

• Kafka 自 2.0.0 版本开始，已经正式摒弃对 Java 7 的支持了，所以有条件的话至少使用 Java 8 吧。

  堆大小参数

• 无脑给出一个通用的建议：将你的 JVM 堆大小设置成 6GB 吧,这是目前业界比较公认的一个合理值。
• 使用默认的 Heap Size 来跑 Kafka，说实话默认的 1GB 有点小，毕竟 Kafka Broker 在与客户端进行交互时会在 JVM 堆上创建大量的 ByteBuffer 实例，Heap Size 不能太小。

  垃圾回收器的设置

• 也就是平时常说的 GC 设置。如果你依然在使用 Java 7，那么可以根据以下法则选择合适的垃圾回收器：
  • 非常充裕，建议使用 CMS 收集器。启用方法是指定-XX:+UseCurrentMarkSweepGC。
  • 否则，使用吞吐量收集器。开启方法是指定-XX:+UseParallelGC。
• 如果你在使用 Java 8，那么可以手动设置使用 G1 收集器。
  • 在没有任何调优的情况下，G1 表现得要比 CMS 出色，主要体现在更少的 Full GC，需要调整的参数更少等，所以使用 G1 就好了。

如何设置JVM参数

• **KAFKA_HEAP_OPTS**：指定堆大小。

• **KAFKA_JVM_PERFORMANCE_OPTS**：指定 GC 参数。

• 启动 Kafka Broker，即在启动 Kafka Broker 之前，先设置上这两个环境变量：

  1 2 3 4

  $> export KAFKA_HEAP_OPTS=--Xms6g --Xmx6g $> export KAFKA_JVM_PERFORMANCE_OPTS= -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true $> bin/kafka-server-start.sh config/server.properties

  操作系统级别的参数

• Kafka 并不需要设置太多的 OS 参数，但有些因素最好还是关注一下，比如下面这几个：
  • 文件描述符限制
  • 文件系统类型
  • Swappiness
  • 提交时间

文件描述符限制

• ulimit -n
  • 任何一个 Java 项目最好都调整下这个值。实际上，文件描述符系统资源并不像我们想象的那样昂贵，你不用太担心调大此值会有什么不利的影响。
  • 通常情况下将它设置成一个超大的值是合理的做法，比如ulimit -n 1000000。
  • 设置这个参数一点都不重要，但不设置的话后果很严重，比如你会经常看到“Too many open files”的错误。

    在 Linux 系统中，一个长连接会占用一个 Socket 句柄（文件描述符），像 Ubuntu 默认是 1024，也就是最多 1024 个 Socket 长连接，Kafka 网络通信中大量使用长连接，这对比较大的 Kafka 集群来说可能是不够的。 为了避免 Socket 句柄不够用，将这个设置为一个比较大值是合理的。

文件系统类型

• 这里所说的文件系统指的是如 ext3、ext4 或 XFS 这样的日志型文件系统。
• 根据官网的测试报告，XFS 的性能要强于 ext4，所以生产环境最好还是使用 XFS。

swap 的调优

• 网上很多文章都提到设置其为 0，将 swap 完全禁掉以防止 Kafka 进程使用 swap 空间。我个人反倒觉得还是不要设置成 0 比较好，我们可以设置成一个较小的值。
• 因为一旦设置成 0，当物理内存耗尽时，操作系统会触发 OOM killer 这个组件，它会随机挑选一个进程然后 kill 掉，即根本不给用户任何的预警。
• 但如果设置成一个比较小的值，当开始使用 swap 空间时，你至少能够观测到 Broker 性能开始出现急剧下降，从而给你进一步调优和诊断问题的时间。
• 基于这个考虑，我个人建议将 swappniess 配置成一个接近 0 但不为 0 的值，比如 1。

提交时间

也称 Flush落盘时间

• 向 Kafka 发送数据并不是真要等数据被写入磁盘才会认为成功，而是只要数据被写入到操作系统的页缓存（Page Cache）上就可以了

• 随后操作系统根据 LRU 算法会定期将页缓存上的“脏”数据落盘到物理磁盘上。

• 这个定期就是由提交时间来确定的，默认是 5 秒。一般情况下我们会认为这个时间太频繁了，可以适当地增加提交间隔来降低物理磁盘的写操作。

• 如果在页缓存中的数据在写入到磁盘前机器宕机了，那岂不是数据就丢失了。的确，这种情况数据确实就丢失了，但鉴于 Kafka 在软件层面已经提供了多副本的冗余机制，因此这里稍微拉大提交间隔去换取性能还是一个合理的做法。

  页缓存属于磁盘缓存（Disk cache）的一种，主要是为了改善系统性能。重复访问磁盘上的磁盘块是常见的操作，把它们保存在内存中可以避免昂贵的磁盘IO操作。 既然叫页缓存，它是根据页（page）来组织的内存结构。每一页包含了很多磁盘上的块数据。Linux使用Radix树实现页缓存，主要是加速特定页的查找速度。另外一般使用LRU策略来淘汰过期页数据。总之它是一个完全由内核来管理的磁盘缓存，用户应用程序通常是无感知的。