Kafka Broker 配置详解

最近在做Kafka集群的副本Leader平衡， 总是有点问题，准备先把我们集群的配置再过一下，再进行下调试。网上写Kafka Broker配置的文章不少，但Kafka本身的配置太多了，真正生产环境不可能都配上，这里挑一些重点的梳理一下， 我们这里以Kafka2.7.2 官方提供的默认模板入手，结合《kafka权威指南》一书的讲解来看，正好它使用的也是Kafka 2.7.0。

Server Basics

• broker.id: 0, broker的整数标识符，在集群中必须唯一

Socket Server Settings

• listeners: 监听器的格式为<protocol>://<hostname>:<port>, 例如PLAINTEXT://localhost:9092， SASL_PLAINTEXT://192.168.1.1:9092等。这里的重点其实是protocol 安全协议，Kafka使用两种标准技术（TLS和 简单身份验证和安全层 SASL simple authentication and security layer）支持4种安全协议，每一个Kafka安全协议都结合了传输层安全（PLAINTEXT或SSL）和可选的认证层安全（SSL或SASL）：

• PLAINTEXT：没有身份验证和加密的PLAINTEXT传输层， 这个是最常见的，开发环境简单搞，就是没有任何加密的裸奔。

• SSL: 带有可选SSL客户端身份验证的SSL传输层，适用于不安全网络，因为它支持客户端和服务器端身份验证以及加密。

• SASL_PLAINTEXT: 带有SASL客户端身份验证的PLAINTEXT传输层。一些SASL机制也支持服务器端身份验证。它不支持加密，因此只适用于内网环境。

• SASL_SSL带有SASL身份验证的SSL传输层，适用于不安全网络，因为它支持客户端和服务器端身份验证以及加密。这个应该是最安全的，外网数据传输都应该走这种协议。

选用PLAINTEXT没有其他参数需要配置，选用其他协议的话都还有些配置需要补充。

• advertised.listeners：将Broker建立通信的地址发布到Zookeeper中，便于Client（Producer和Consumer）连接。如果不配置的话会默认使用listeners的配置。

• listener.security.protocol.map：以Key/Value的形式定义监听者的安全协议，在大多数情况下会将Key认为是监听者的别名。这个可以不需要单独配置，直接使用默认值即可：

listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

Log Basics

• log.dirs：Message保存的路径。

  
  

• num.partitions：创建Topic时，如果没有指定Partition数量，则使用该配置项设置的Partition数量。

  
  

• num.recovery.threads.per.data.dir：Kafka使用线程池来处理日志片段，每个目录使用的线程数量。目前，线程池被用于以下3种情形：

• 当服务器正常启动时，用于打开每个分区的日志片段。

•  当服务器发生崩溃并重启时，用于检查和截断每个分区的日志片段。

•  当服务器正常关闭时，用于关闭日志片段。

默认值为1，因为这些线程只在服务器启动和关闭时使用，所以可以多设置一些线程来实现并行操作。特别是对包含大量分区的服务器来说，一旦发生崩溃，在从错误中恢复时可以通过并行操作省下很多时间。

Internal Topic Settings

以下几个都是内部的metadata topic（__consumer_offsets 和 __transaction_state）的 副本集配置，默认值为1，生产环境起码调整为3

offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3

Log Retention Policy

• log.retention.hours: 用于确定消息将在多久以后被删除，还有另外两个参数log.retention.minutes和log.retention.ms。这3个参数的作用是一样的。

  
  

• log.retention.bytes：已保留的消息的字节总数来判断旧消息是否过期，这个值对应的是每一个分区，如果一个topic有3个分区，这个值设为1GB, 那么该主题最多保留3GB数据。

如果同时指定了log.retention.bytes和log.retention.hours，那么只要任意一个条件得到满足，消息就会被删除。为简单起见，建议只选择其中的一种保留策略，要么基于数据大小，要么基于时间，或者两种都不选择，以防发生意外的数据丢失。

• log.segment.bytes: 当消息到达broker时，它们会被追加到分区的当前日志片段上。当日志片段大小达到log.segment.bytes指定的上限（默认是1 GB）时，当前日志片段会被关闭，一个新的日志片段会被打开。如果主题的消息量不是很大，那么如何设置这个参数就变得尤为重要。如果一个主题每天只接收100 MB的消息，并且log.segment.bytes使用了默认设置，那么填满一个日志片段将需要10天, 在日志片段被关闭之前消息是不会过期的, 如果log.retention.hours设置为1天的话，那么日志片段11天才会过期。

Zookeeper

2.7版本还在使用zk，3.X的版本已经干掉了它。

• zookeeper.connect：用于保存broker元数据的ZooKeeper地址，我们常见的是只指定hostname和端口，其实后面是可以带上path的作为Kafka集群的chroot，hostname:port/path， 不指定的话，默认是使用的根路径，带上路径的话更方便于其他应用共享zk，比如ck等。

Group Coordinator Settings

• group.initial.rebalance.delay.ms：当Consumer Group新增或减少Consumer时，重新分配Topic Partition的延迟时间。这里官方给出的默认模板给的是0，这是有个小坑的，这个配置只适合于测试，生产环境最好使用3s：

# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.

以上基本上覆盖了Kafka提供默认模板中的配置，其他还有些默认配置，我们也需要在生产环境中按需修改

• message.max.bytes：限制单条消息的大小，默认值是1 000000，也就是1 MB。如果生产者尝试发送超过这个大小的消息，那么不仅消息不会被broker接收，还会收到broker返回的错误信息。与其他broker配置参数一样，这个参数指的是压缩后的消息大小，也就是说，消息的实际大小可以远大于message.max.bytes，只要压缩后小于这个值即可。这个配置其实和生产者和消费者的配置都高度相关，其中任何一个环节配置过小都可能导致大的消息拿不到，以后找机会单独写一下。

• auto.leader.rebalance.enable：是否开启leader自动平衡，默认值为true。为了确保主题的所有权不会集中在一台broker上，可以将这个参数设置为true，让主题的所有权尽可能地在集群中保持均衡。如果启用了这个功能，那么就会有一个后台线程定期检查分区的分布情况（这个时间间隔可以通过leader.imbalance.check.interval.seconds来配置）。如果不均衡的所有权超出了leader.imbalance.per.broker.percentage指定的百分比，则会启动一次分区首领再均衡。这个参数网上有的文章建议是不开启， 原因是线上环境的自动迁移副本还是有风险的。kafka权威指南并没有建议不开启，但我个人认为，在自身数据量不大，topic数量可控的情况下，还是可以开启的，这样便于数据的自动平衡。

小结

一个简单的不带安全认证的kafka的server.properties配置大致如下：

############################# Server Basics #############################
broker.id=1


############################# Socket Server Settings #############################
listeners=PLAINTEXT://kafka1:9092


advertised.listeners=PLAINTEXT://kafka1:9092


num.network.threads=3


num.io.threads=8


socket.send.buffer.bytes=102400


socket.receive.buffer.bytes=102400


socket.request.max.bytes=104857600




############################# Log Basics #############################
log.dirs=/xxx/xxx


num.partitions=3


num.recovery.threads.per.data.dir=4


############################# Internal Topic Settings  #############################
offsets.topic.replication.factor=3


transaction.state.log.replication.factor=3


transaction.state.log.min.isr=1




############################# Log Retention Policy #############################


log.retention.hours=24


log.segment.bytes=1073741824


log.retention.check.interval.ms=300000


############################# Zookeeper #############################


zookeeper.connect=zk1:2181,zk2:2181,zk3:2181


zookeeper.connection.timeout.ms=60000


############################# Group Coordinator Settings #############################
group.initial.rebalance.delay.ms=30000


############################# Other #############################
message.max.bytes: 10000000

Kafka的配置非常多，这里还只是覆盖了一小部分，kafka提供的默认模板只能作为测试环境的配置使用，生产环境很多参数都还需要调优。现在我们的团队在生产部署了不少大数据服务的组件， 但很多都使用的一些未进行调优的默认配置。随着我们承接的数据量也越来越大了，同时我们的服务器逐步在老化，故障越来越多，这些不合理的配置在服务器故障和数据量的冲击下逐步暴露出各种问题，这个时候就得静下心来，花时间去深钻这些原理了。