Kafka Broker配置（0.10版）

系统管理员 2021-06-24 15:59 282阅读 0赞

官网：[https://kafka.apache.org/documentation/][https_kafka.apache.org_documentation]

其余配置可参考原作者其他翻译文档：[https://www.orchome.com/535][https_www.orchome.com_535]

基本配置如下:

*  broker.id
 *  log.dirs
 *  zookeeper.connect

下文将详细论述了主题级别配置和默认值。

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>描述</th> 
   <th>类型</th> 
   <th>默认</th> 
   <th>有效值</th> 
   <th>重要程度</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>zookeeper.connect</td> 
   <td>zookeeper host string</td> 
   <td>string</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>advertised.host.name</td> 
   <td>过时的：当<code>advertised.listeners</code>或<code>listeners</code>没设置时候才使用。请改用<code>advertised.listeners</code>。Hostname发布到Zookeeper供客户端使用。在IaaS环境中，Broker可能需要绑定不同的接口。如果没有设置，将会使用<code>host.name</code>（如果配置了）。否则将从<code>java.net.InetAddress.getCanonicalHostName()</code>获取。</td> 
   <td>string</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>advertised.listeners</td> 
   <td>发布到Zookeeper供客户端使用监听（如果不同）。在IaaS环境中，broker可能需要绑定不同的接口。如果没设置，则使用<code>listeners</code>。</td> 
   <td>string</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>advertised.port</td> 
   <td>过时的：当<code>advertised.listeners</code>或<code>listeners</code>没有设置才使用。请改用<code>advertised.listeners</code>。端口发布到Zookeeper供客户端使用，在IaaS环境中，broker可能需要绑定到不同的端口。如果没有设置，将和broker绑定的同一个端口。</td> 
   <td>int</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>auto.create.topics.enable</td> 
   <td>启用自动创建topic</td> 
   <td>boolean</td> 
   <td>true</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>auto.leader.rebalance.enable</td> 
   <td>启用自动平衡leader。如果需要，后台线程会定期检查并触发leader平衡。</td> 
   <td>boolean</td> 
   <td>true</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>background.threads</td> 
   <td>用于各种后台处理任务的线程数</td> 
   <td>int</td> 
   <td>10</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>broker.id</td> 
   <td>服务器的broker id。如果未设置，将生成一个独一无二的broker id。要避免zookeeper生成的broker id和用户配置的broker id冲突，从reserved.broker.max.id + 1开始生成。</td> 
   <td>int</td> 
   <td>-1</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>compression.type</td> 
   <td>为给定topic指定最终的压缩类型。支持标准的压缩编码器（'gzip', 'snappy', 'lz4'）。也接受'未压缩'，就是没有压缩。保留由producer设置的原始的压缩编码。</td> 
   <td>string</td> 
   <td>producer</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>delete.topic.enable</td> 
   <td>启用删除topic。如果此配置已关闭，通过管理工具删除topic将没有任何效果</td> 
   <td>boolean</td> 
   <td>false</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>host.name</td> 
   <td>不赞成：当<code>listeners</code>没有设置才会使用。请改用<code>listeners</code>。如果设置，它将只绑定到此地址。如果没有设置，它将绑定到所有接口</td> 
   <td>string</td> 
   <td>""</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>leader.imbalance.check.interval.seconds</td> 
   <td>由控制器触发分区再平衡检查的频率</td> 
   <td>long</td> 
   <td>300</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>leader.imbalance.per.broker.percentage</td> 
   <td>允许每个broker的leader比例不平衡。如果每个broker的值高于此值，控制器将触发leader平衡，该值以百分比的形式指定。</td> 
   <td>int</td> 
   <td>10</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>listeners</td> 
   <td>监听列表 - 监听逗号分隔的URL列表和协议。指定hostname为0.0.0.0绑定到所有接口，将hostname留空则绑定到默认接口。合法的listener列表是：PLAINTEXT://myhost:9092,TRACE://:9091 PLAINTEXT://0.0.0.0:9092, TRACE://localhost:9093</td> 
   <td>string</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.dir</td> 
   <td>保存日志数据的目录 (补充log.dirs属性)</td> 
   <td>string</td> 
   <td>/tmp/kafka-logs</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.dirs</td> 
   <td>保存日志数据的目录。如果未设置，则使用log.dir中的值</td> 
   <td>string</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.flush.interval.messages</td> 
   <td>消息刷新到磁盘之前，累计在日志分区的消息数</td> 
   <td>long</td> 
   <td>9223372036854775807</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.flush.interval.ms</td> 
   <td>topic中的消息在刷新到磁盘之前保存在内存中的最大时间（以毫秒为单位），如果未设置，则使用log.flush.scheduler.interval.ms中的值</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.flush.offset.checkpoint.interval.ms</td> 
   <td>我们更新的持续记录的最后一次刷新的频率。作为日志的恢复点。</td> 
   <td>int</td> 
   <td>60000</td> 
   <td>[0,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.flush.scheduler.interval.ms</td> 
   <td>日志刷新的频率（以毫秒为单位）检查是否有任何日志需要刷新到磁盘</td> 
   <td>long</td> 
   <td>9223372036854775807</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.retention.bytes</td> 
   <td>删除日志之前的最大大小</td> 
   <td>long</td> 
   <td>-1</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.retention.hours</td> 
   <td>删除日志文件保留的小时数（以小时为单位）。第三级是log.retention.ms属性</td> 
   <td>int</td> 
   <td>168</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.retention.minutes</td> 
   <td>删除日志文件之前保留的分钟数（以分钟为单位）。次于log.retention.ms属性。如果没设置，则使用log.retention.hours的值。</td> 
   <td>int</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.retention.ms</td> 
   <td>删除日志文件之前保留的毫秒数（以毫秒为单位），如果未设置，则使用log.retention.minutes的值。</td> 
   <td>long</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.roll.hours</td> 
   <td>新建一个日志段的最大时间（以小时为单位），次于log.roll.ms属性。</td> 
   <td>int</td> 
   <td>168</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.roll.jitter.hours</td> 
   <td>从logRollTimeMillis（以小时为单位）减去最大抖动，次于log.roll.jitter.ms属性。</td> 
   <td>int</td> 
   <td>0</td> 
   <td>[0,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.roll.ms</td> 
   <td>新建一个日志段之前的最大事时间（以毫秒为单位）。如果未设置，则使用log.roll.hours的值。</td> 
   <td>long</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.segment.bytes</td> 
   <td>单个日志文件的最大大小</td> 
   <td>int</td> 
   <td>1073741824</td> 
   <td>[14,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>log.segment.delete.delay.ms</td> 
   <td>从文件系统中删除文件之前的等待的时间</td> 
   <td>long</td> 
   <td>60000</td> 
   <td>[0,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>message.max.bytes</td> 
   <td>服务器可以接收的消息的最大大小</td> 
   <td>int</td> 
   <td>1000012</td> 
   <td>[0,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>min.insync.replicas</td> 
   <td>当producer设置acks为"all"（或"-1"）时。min.insync.replicas指定必须应答成功写入的replicas最小数。如果不能满足最小值，那么producer抛出一个异常（NotEnoughReplicas或NotEnoughReplicasAfterAppend）。<br> 当一起使用时，min.insync.replicas和acks提供最大的耐用性保证。一个典型的场景是创建一个复制因子3的topic，设置min.insync.replicas为2，并且ack是“all”。如果多数副本没有接到写入时，将会抛出一个异常。</td> 
   <td>int</td> 
   <td>1</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>num.io.threads</td> 
   <td>服务器用于执行网络请求的io线程数</td> 
   <td>int</td> 
   <td>8</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>num.network.threads</td> 
   <td>服务器用于处理网络请求的线程数。</td> 
   <td>int</td> 
   <td>3</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>num.recovery.threads.per.data.dir</td> 
   <td>每个数据的目录线程数，用于启动时日志恢复和关闭时flush。</td> 
   <td>int</td> 
   <td>1</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>num.replica.fetchers</td> 
   <td>从源broker复制消息的提取线程数。递增该值可提高follower broker的I/O的并发。</td> 
   <td>int</td> 
   <td>1</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>offset.metadata.max.bytes</td> 
   <td>offset提交关联元数据条目的最大大小</td> 
   <td>int</td> 
   <td>4096</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>offsets.commit.required.acks</td> 
   <td>commit之前需要的应答数，通常，不应覆盖默认的（-1）</td> 
   <td>short</td> 
   <td>-1</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>offsets.commit.timeout.ms</td> 
   <td>Offset提交延迟，直到所有副本都收到提交或超时。 这类似于生产者请求超时。</td> 
   <td>int</td> 
   <td>5000</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>offsets.load.buffer.size</td> 
   <td>当加载offset到缓存时，从offset段读取的批量大小。</td> 
   <td>int</td> 
   <td>5242880</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>offsets.retention.check.interval.ms</td> 
   <td>检查过期的offset的频率。</td> 
   <td>long</td> 
   <td>600000</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>offsets.retention.minutes</td> 
   <td>offset topic的日志保留时间（分钟）</td> 
   <td>int</td> 
   <td>1440</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>offsets.topic.compression.codec</td> 
   <td>压缩编码器的offset topic - 压缩可以用于实现“原子”提交</td> 
   <td>int</td> 
   <td>0</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>offsets.topic.num.partitions</td> 
   <td>offset commit topic的分区数（部署之后不应更改）</td> 
   <td>int</td> 
   <td>50</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>offsets.topic.replication.factor</td> 
   <td>offset topic复制因子（ps：就是备份数，设置的越高来确保可用性）。为了确保offset topic有效的复制因子，第一次请求offset topic时，活的broker的数量必须最少最少是配置的复制因子数。 如果不是，offset topic将创建失败或获取最小的复制因子（活着的broker，复制因子的配置）</td> 
   <td>short</td> 
   <td>3</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>offsets.topic.segment.bytes</td> 
   <td>offset topic段字节应该相对较小一点，以便于加快日志压缩和缓存加载</td> 
   <td>int</td> 
   <td>104857600</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>port</td> 
   <td>不赞成：当<code>listener</code>没有设置才使用。请改用<code>listeners</code>。该port监听和接收连接。</td> 
   <td>int</td> 
   <td>9092</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>queued.max.requests</td> 
   <td>在阻塞网络线程之前允许的排队请求数</td> 
   <td>int</td> 
   <td>500</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>quota.consumer.default</td> 
   <td>过时的：当默认动态的quotas没有配置或在Zookeeper时。如果每秒获取的字节比此值高，所有消费者将通过clientId/consumer区分限流。</td> 
   <td>long</td> 
   <td>9223372036854775807</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>quota.producer.default</td> 
   <td>过时的：当默认动态的quotas没有配置，或在zookeeper时。如果生产者每秒比此值高，所有生产者将通过clientId区分限流。</td> 
   <td>long</td> 
   <td>9223372036854775807</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>replica.fetch.min.bytes Minimum</td> 
   <td>每个获取响应的字节数。如果没有满足字节数，等待replicaMaxWaitTimeMs。</td> 
   <td>int</td> 
   <td>1</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>replica.fetch.wait.max.ms</td> 
   <td>跟随者副本发出每个获取请求的最大等待时间，此值应始终小于replica.lag.time.max.ms，以防止低吞吐的topic的ISR频繁的收缩。</td> 
   <td>int</td> 
   <td>500</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>replica.high.watermark.<br> checkpoint.interval.ms</td> 
   <td>达到高“水位”保存到磁盘的频率。</td> 
   <td>long</td> 
   <td>5000</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>replica.lag.time.max.ms</td> 
   <td>如果一个追随者没有发送任何获取请求或至少在这个时间的这个leader的没有消费完。该leader将从isr中移除这个追随者。</td> 
   <td>long</td> 
   <td>10000</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>replica.socket.receive.buffer.bytes</td> 
   <td>用于网络请求的socket接收缓存区</td> 
   <td>int</td> 
   <td>65536</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>replica.socket.timeout.ms</td> 
   <td>网络请求的socket超时，该值最少是replica.fetch.wait.max.ms</td> 
   <td>int</td> 
   <td>30000</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>request.timeout.ms</td> 
   <td>该配置控制客户端等待请求的响应的最大时间，。如果超过时间还没收到消费。客户端将重新发送请求，如果重试次数耗尽，则请求失败。</td> 
   <td>int</td> 
   <td>30000</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>socket.receive.buffer.bytes</td> 
   <td>socket服务的SO_RCVBUF缓冲区。如果是-1，则默认使用OS的。</td> 
   <td>int</td> 
   <td>102400</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>socket.request.max.bytes</td> 
   <td>socket请求的最大字节数</td> 
   <td>int</td> 
   <td>104857600</td> 
   <td>[1,...]</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>socket.send.buffer.bytes</td> 
   <td>socket服务的SO_SNDBUF缓冲区。如果是-1，则默认使用OS的。</td> 
   <td>int</td> 
   <td>102400</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>unclean.leader.election.enable</td> 
   <td>是否启用不在ISR中的副本参与选举leader的最后的手段。这样做有可能丢失数据。</td> 
   <td>boolean</td> 
   <td>false</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>zookeeper.connection.timeout.ms</td> 
   <td>连接zookeeper的最大等待时间，如果未设置，则使用zookeeper.session.timeout.ms。</td> 
   <td>int</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>zookeeper.session.timeout.ms</td> 
   <td>Zookeeper会话的超时时间</td> 
   <td>int</td> 
   <td>6000</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>zookeeper.set.acl</td> 
   <td>设置客户端使用安全的ACL</td> 
   <td>boolean</td> 
   <td>false</td> 
   <td>&nbsp;</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>broker.id.generation.enable</td> 
   <td>启用自动生成broker id。启用该配置时应检查reserved.broker.max.id。</td> 
   <td>boolean</td> 
   <td>true</td> 
   <td>&nbsp;</td> 
   <td>中等</td> 
  </tr> 
  <tr> 
   <td>broker.rack</td> 
   <td>broker机架，用于机架感知副本分配的失败容错。例如：<code>RACK1</code>, <code>us-east-1d</code></td> 
   <td>string</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>中等</td> 
  </tr> 
  <tr> 
   <td>connections.max.idle.ms Idle</td> 
   <td>连接超时：闲置时间超过该设置，则服务器socket处理线程关闭这个连接。</td> 
   <td>long</td> 
   <td>600000</td> 
   <td>&nbsp;</td> 
   <td>中等</td> 
  </tr> 
  <tr> 
   <td>controlled.shutdown.enable</td> 
   <td>启用服务器的关闭控制。</td> 
   <td>boolean</td> 
   <td>true</td> 
   <td>&nbsp;</td> 
   <td>中等</td> 
  </tr> 
  <tr> 
   <td>controlled.shutdown.max.retries</td> 
   <td>控制因多种原因导致的shutdown失败，当这样失败发生，尝试重试的次数</td> 
   <td>int</td> 
   <td>3</td> 
   <td>&nbsp;</td> 
   <td>中等</td> 
  </tr> 
  <tr> 
   <td>controlled.shutdown.retry.backoff.ms</td> 
   <td>在每次重试之前，系统需要时间从导致先前故障的状态（控制器故障转移，复制延迟等）恢复。 此配置是重试之前等待的时间数。</td> 
   <td>long</td> 
   <td>5000</td> 
   <td>&nbsp;</td> 
   <td>中等</td> 
  </tr> 
  <tr> 
   <td>controller.socket.timeout.ms</td> 
   <td>控制器到broker通道的sockt超时时间</td> 
   <td>int</td> 
   <td>30000</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>default.replication.factor</td> 
   <td>自动创建topic的默认的副本数</td> 
   <td>int</td> 
   <td>1</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>fetch.purgatory.purge.interval.requests</td> 
   <td>拉取请求清洗间隔（请求数）</td> 
   <td>int</td> 
   <td>1000</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>group.max.session.timeout.ms</td> 
   <td>已注册的消费者允许的最大会话超时时间，设置的时候越长使消费者有更多时间去处理心跳之间的消息。但察觉故障的时间也拉长了。</td> 
   <td>int</td> 
   <td>300000</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>group.min.session.timeout.ms</td> 
   <td>已经注册的消费者允许最小的会话超时时间，更短的时间去快速的察觉到故障，代价是频繁的心跳，这可能会占用大量的broker资源。</td> 
   <td>int</td> 
   <td>6000</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>inter.broker.protocol.version</td> 
   <td>指定broker内部通讯使用的版本。通常在更新broker时使用。有效的值为：0.8.0, 0.8.1, 0.8.1.1, 0.8.2, 0.8.2.0, 0.8.2.1, 0.9.0.0, 0.9.0.1。查看ApiVersion找到的全部列表。</td> 
   <td>string</td> 
   <td>0.10.1-IV2</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.cleaner.backoff.ms</td> 
   <td>当没有日志要清理时，休眠的时间</td> 
   <td>long</td> 
   <td>15000</td> 
   <td>[0,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.cleaner.dedupe.buffer.size</td> 
   <td>用于日志去重的内存总量（所有cleaner线程）</td> 
   <td>long</td> 
   <td>134217728</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.cleaner.delete.retention.ms</td> 
   <td>删除记录保留多长时间？</td> 
   <td>long</td> 
   <td>86400000</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.cleaner.enable</td> 
   <td>在服务器上启用日志清洗处理？如果使用的任何topic的cleanup.policy=compact包含内部的offset topic，应启动。如果禁用，那些topic将不会被压缩并且会不断的增大。</td> 
   <td>boolean</td> 
   <td>true</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.cleaner.io.buffer.load.factor</td> 
   <td>日志cleaner去重缓冲负载因子。去重缓冲区的百分比，较高的值将允许同时清除更多的日志，但将会导致更多的hash冲突。</td> 
   <td>double</td> 
   <td>0.9</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.cleaner.io.buffer.size</td> 
   <td>所有日志清洁器线程I/O缓存的总内存</td> 
   <td>int</td> 
   <td>524288</td> 
   <td>[0,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.cleaner.io.max.bytes.per.second</td> 
   <td>日志清理器限制，以便其读写i/o平均小与此值。</td> 
   <td>double</td> 
   <td>1.7976931348623157E308</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.cleaner.min.cleanable.ratio</td> 
   <td>脏日志与日志的总量的最小比率，以符合清理条件</td> 
   <td>double</td> 
   <td>0.5</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.cleaner.min.compaction.lag.ms</td> 
   <td>一条消息在日志保留不压缩的最小时间，仅适用于正在压缩的日志。</td> 
   <td>long</td> 
   <td>0</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.cleaner.threads</td> 
   <td>用于日志清除的后台线程数</td> 
   <td>int</td> 
   <td>1</td> 
   <td>[0,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.cleanup.policy</td> 
   <td>超过保留时间段的默认清除策略。逗号分隔的有效的策略列表。有效的策略有：“delete”和“compact”</td> 
   <td>list</td> 
   <td>[delete]</td> 
   <td>[compact, delete]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.index.interval.bytes</td> 
   <td>添加一个条目到offset的间隔</td> 
   <td>index</td> 
   <td>int</td> 
   <td>4096</td> 
   <td>[0,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.index.size.max.bytes</td> 
   <td>offset index的最大大小（字节）</td> 
   <td>int</td> 
   <td>10485760</td> 
   <td>[4,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.message.format.version</td> 
   <td>指定追加到日志中的消息格式版本。例如： 0.8.2, 0.9.0.0, 0.10.0。通过设置一个特定消息格式版本，用户需要保证磁盘上所有现有的消息小于或等于指定的版本。错误的设置将导致旧版本的消费者中断，因为消费者接收一个不理解的消息格式。</td> 
   <td>string</td> 
   <td>0.10.1-IV2</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.message.timestamp.difference.max.ms</td> 
   <td>如果log.message.timestamp.type=CreateTime，broker接收消息时的时间戳和消息中指定的时间戳之间允许的最大差异。如果时间戳超过此阈值，则消息将被拒绝。如果log.message.timestamp.type=LogAppendTime，则此配置忽略。</td> 
   <td>long</td> 
   <td>9223372036854775807</td> 
   <td>[0,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.message.timestamp.type</td> 
   <td>定义消息中的时间戳是消息创建时间或日志追加时间。该值可设置为<code>CreateTime</code> 或 <code>LogAppendTime</code></td> 
   <td>string</td> 
   <td>CreateTime</td> 
   <td>[CreateTime, LogAppendTime]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.preallocate</td> 
   <td>在创建新段时预分配文件？如果你在Windowns上使用kafka，你可能需要设置它为true。</td> 
   <td>boolean</td> 
   <td>false</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>log.retention.check.interval.ms</td> 
   <td>日志清除程序检查日志是否满足被删除的频率（以毫秒为单位）</td> 
   <td>long</td> 
   <td>300000</td> 
   <td>[1,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>max.connections.per.ip</td> 
   <td>允许每个ip地址的最大连接数。</td> 
   <td>int</td> 
   <td>2147483647</td> 
   <td>[1,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>max.connections.per.ip.overrides</td> 
   <td>per-ip或hostname覆盖默认最大连接数</td> 
   <td>string</td> 
   <td>""</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>num.partitions</td> 
   <td>topic的默认分区数</td> 
   <td>int</td> 
   <td>1</td> 
   <td>[1,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>principal.builder.class</td> 
   <td>实现PrincipalBuilder接口类的完全限定名，该接口目前用于构建与SSL SecurityProtocol连接的Principal。</td> 
   <td>class</td> 
   <td>class org.apache.kafka.<br> common.security.auth<br> .DefaultPrincipalBuilder</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>producer.purgatory.purge.interval.requests</td> 
   <td>生产者请求purgatory的清洗间隔（请求数）</td> 
   <td>int</td> 
   <td>1000</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>replica.fetch.backoff.ms</td> 
   <td>当拉取分区发生错误时休眠的时间</td> 
   <td>1000</td> 
   <td>[0,...]</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>replica.fetch.max.bytes</td> 
   <td>拉取每个分区的消息的字节数。这不是绝对的最大值，如果提取的第一个非空分区中的第一个消息大于这个值，则消息仍然返回，以确保进展。通过message.max.bytes (broker配置)或max.message.bytes (topic配置)定义broker接收的最大消息大小。</td> 
   <td>int</td> 
   <td>1048576</td> 
   <td>[0,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>replica.fetch.response.max.bytes</td> 
   <td>预计整个获取响应的最大字节数，这不是绝对的最大值，如果提取的第一个非空分区中的第一个消息大于这个值，则消息仍然返回，以确保进展。通过message.max.bytes (broker配置)或max.message.bytes (topic配置)定义broker接收的最大消息大小。</td> 
   <td>int</td> 
   <td>10485760</td> 
   <td>[0,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>reserved.broker.max.id</td> 
   <td>broker.id的最大数</td> 
   <td>int</td> 
   <td>1000</td> 
   <td>[0,...]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>sasl.enabled.mechanisms</td> 
   <td>可用的SASL机制列表，包含任何可用的安全提供程序的机制。默认情况下只有GSSAPI是启用的。</td> 
   <td>list</td> 
   <td>[GSSAPI]</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.kinit.cmd</td> 
   <td>Kerberos kinit 命令路径。</td> 
   <td>string</td> 
   <td>/usr/bin/kinit</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.min.time.before.relogin</td> 
   <td>登录线程在刷新尝试的休眠时间。</td> 
   <td>long</td> 
   <td>60000</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.principal.to.local.rules</td> 
   <td>principal名称映射到一个短名称（通常是操作系统用户名）。按顺序，使用与principal名称匹配的第一个规则将其映射其到短名称。忽略后面的规则。默认情况下，{username}/{hostname}@{REALM} 映射到 {username}。</td> 
   <td>list</td> 
   <td>[DEFAULT]</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.service.name</td> 
   <td>Kafka运行的Kerberos principal名称。 可以在JAAS或Kafka的配置文件中定义。</td> 
   <td>string</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.ticket.renew.jitter</td> 
   <td>添加到更新时间的随机抖动的百分比</td> 
   <td>time. double</td> 
   <td>0.05</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.ticket.renew.window.factor</td> 
   <td>登录线程休眠，直到从上次刷新到ticket的到期的时间已到达（指定窗口因子），在此期间它将尝试更新ticket。</td> 
   <td>double</td> 
   <td>0.8</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>sasl.mechanism.inter.broker.protocol</td> 
   <td>SASL机制，用于broker之间的通讯，默认是GSSAPI。</td> 
   <td>string</td> 
   <td>GSSAPI</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>security.inter.broker.protocolSecurity</td> 
   <td>broker之间的通讯协议，有效值有：PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL。</td> 
   <td>string</td> 
   <td>PLAINTEXT</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.cipher.suites</td> 
   <td>密码套件列表。认证，加密，MAC和秘钥交换算法的组合，用于使用TLS或SSL的网络协议<code>交涉</code>网络连接的安全设置，默认情况下，支持所有可用的密码套件。</td> 
   <td>list</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.client.auth</td> 
   <td>配置请求客户端的broker认证。常见的设置：<br><em><code>ssl.client.auth=required</code> 需要客户端认证。</em><br><code>ssl.client.auth=requested</code> 客户端认证可选，不同于requested ，客户端可选择不提供自身的身份验证信息<br> * <code>ssl.client.auth=none</code> 不需要客户端身份认证</td> 
   <td>string</td> 
   <td>none</td> 
   <td>[required, requested, none]</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.enabled.protocols</td> 
   <td>已启用的SSL连接协议列表。</td> 
   <td>list</td> 
   <td>[TLSv1.2, TLSv1.1, TLSv1]</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.key.password</td> 
   <td>秘钥库文件中的私钥密码。对客户端是可选的。</td> 
   <td>password</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.keymanager.algorithm</td> 
   <td>用于SSL连接的密钥管理工厂算法。默认值是Java虚拟机的密钥管理工厂算法。</td> 
   <td>string</td> 
   <td>SunX509</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.keystore.location</td> 
   <td>密钥仓库文件的位置。客户端可选，并可用于客户端的双向认证。</td> 
   <td>string</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.keystore.password</td> 
   <td>密钥仓库文件的仓库密码。客户端可选，只有ssl.keystore.location配置了才需要。</td> 
   <td>password</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.keystore.type</td> 
   <td>密钥仓库文件的格式。客户端可选。</td> 
   <td>string</td> 
   <td>JKS</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.protocol</td> 
   <td>用于生成SSLContext，默认是TLS，适用于大多数情况。允许使用最新的JVM，LS, TLSv1.1 和TLSv1.2。 SSL，SSLv2和SSLv3 老的JVM也可能支持，由于有已知的安全漏洞，不建议使用。</td> 
   <td>string</td> 
   <td>TLS</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.provider</td> 
   <td>用于SSL连接的安全提供程序的名称。默认值是JVM的安全程序。</td> 
   <td>string</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.trustmanager.algorithm</td> 
   <td>信任管理工厂用于SSL连接的算法。默认为Java虚拟机配置的信任算法。</td> 
   <td>string</td> 
   <td>PKIX</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.truststore.location</td> 
   <td>信任仓库文件的位置</td> 
   <td>string</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.truststore.password</td> 
   <td>信任仓库文件的密码</td> 
   <td>password</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>ssl.truststore.type</td> 
   <td>信任仓库文件的文件格式</td> 
   <td>string</td> 
   <td>JKS</td> 
   <td>&nbsp;</td> 
   <td>中</td> 
  </tr> 
  <tr> 
   <td>authorizer.class.name</td> 
   <td>用于认证的授权程序类</td> 
   <td>string</td> 
   <td>""</td> 
   <td>&nbsp;</td> 
   <td>低</td> 
  </tr> 
  <tr> 
   <td>metric.reporters</td> 
   <td>度量报告的类列表，通过实现<code>MetricReporter</code>接口，允许插入新度量标准类。JmxReporter包含注册JVM统计。</td> 
   <td>list</td> 
   <td>[]</td> 
   <td>&nbsp;</td> 
   <td>低</td> 
  </tr> 
  <tr> 
   <td>metrics.num.samples</td> 
   <td>维持计算度量的样本数。</td> 
   <td>int</td> 
   <td>2</td> 
   <td>[1,...]</td> 
   <td>低</td> 
  </tr> 
  <tr> 
   <td>metrics.sample.window.ms</td> 
   <td>计算度量样本的时间窗口</td> 
   <td>long</td> 
   <td>30000</td> 
   <td>[1,...]</td> 
   <td>低</td> 
  </tr> 
  <tr> 
   <td>quota.window.num</td> 
   <td>在内存中保留客户端限额的样本数</td> 
   <td>int</td> 
   <td>11</td> 
   <td>[1,...]</td> 
   <td>低</td> 
  </tr> 
  <tr> 
   <td>quota.window.size.seconds</td> 
   <td>每个客户端限额的样本时间跨度</td> 
   <td>int</td> 
   <td>1</td> 
   <td>[1,...]</td> 
   <td>低</td> 
  </tr> 
  <tr> 
   <td>replication.quota.window.num</td> 
   <td>在内存中保留副本限额的样本数</td> 
   <td>int</td> 
   <td>11</td> 
   <td>[1,...]</td> 
   <td>低</td> 
  </tr> 
  <tr> 
   <td>replication.quota.window.size.seconds</td> 
   <td>每个副本限额样本数的时间跨度</td> 
   <td>int</td> 
   <td>1</td> 
   <td>[1,...]</td> 
   <td>低</td> 
  </tr> 
  <tr> 
   <td>ssl.endpoint.identification.algorithm</td> 
   <td>端点身份标识算法，使用服务器证书验证服务器主机名。</td> 
   <td>string</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>低</td> 
  </tr> 
  <tr> 
   <td>ssl.secure.random.implementation</td> 
   <td>用于SSL加密操作的SecureRandom PRNG实现。</td> 
   <td>string</td> 
   <td>null</td> 
   <td>&nbsp;</td> 
   <td>低</td> 
  </tr> 
  <tr> 
   <td>zookeeper.sync.time.ms</td> 
   <td>ZK follower可落后与leader多久。</td> 
   <td>int</td> 
   <td>2000</td> 
   <td>&nbsp;</td> 
   <td>低</td> 
  </tr> 
 </tbody> 
</table>

更多关于broker配置的详情，可以在scala类中的`kafka.server.KafkaConfig`找到。

[https_kafka.apache.org_documentation]: https://kafka.apache.org/documentation/
[https_www.orchome.com_535]: https://www.orchome.com/535