librdkafka配置参数

我会带着你远行 2022-01-15 15:43 290阅读 0赞

http://kafka.apache.org/documentation.html\#configuration

## 全局配置属性 ##

<table> 
 <thead> 
 <tr> 
 <th>属性</th> 
 <th>C/P</th> 
 <th>范围</th> 
 <th style="width:122px;">默认值</th> 
 <th style="width:392px;">描述</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>builtin.features</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;"> gzip, snappy, ssl, sasl, regex, lz4 </td> 
 <td style="width:392px;">标示该librdkafka的支持的内建特性。应用程序可以查看或设置这些值来检查是否支持这些特性。 Type: CSV flags</td> 
 </tr> 
 <tr> 
 <td>client.id</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">rdkafka</td> 
 <td style="width:392px;">客户端标示。Type: string</td> 
 </tr> 
 <tr> 
 <td>metadata.broker.list</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">初始化的broker列表。应用程序也可以使用&nbsp;<code>rd_kafka_brokers_add()</code>&nbsp;在运行时添加 broker。 Type: string</td> 
 </tr> 
 <tr> 
 <td>bootstrap.servers</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">参考&nbsp;<code>metadata.broker.list</code></td> 
 </tr> 
 <tr> 
 <td>message.max.bytes</td> 
 <td>*</td> 
 <td>1000 .. 1000000000</td> 
 <td style="width:122px;">1000000</td> 
 <td style="width:392px;">最大发送消息大小。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>message.copy.max.bytes</td> 
 <td>*</td> 
 <td>0 .. 1000000000</td> 
 <td style="width:122px;">65535</td> 
 <td style="width:392px;">消息拷贝到缓存的最大大小。如果消息大于这个值，将会消耗更多的iovec而采用引用（零拷贝）方式。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>receive.message.max.bytes</td> 
 <td>*</td> 
 <td>1000 .. 1000000000</td> 
 <td style="width:122px;">100000000</td> 
 <td style="width:392px;">最大接收消息大小。这是一个安全预防措施，防止协议饱和时内存耗尽。这个值至少为 fetch.message.max.bytes * 消费者分区数 + 消息头大小 (e.g. 200000 bytes).&nbsp; Type: integer</td> 
 </tr> 
 <tr> 
 <td>max.in.flight.requests.per.connection</td> 
 <td>*</td> 
 <td>1 .. 1000000</td> 
 <td style="width:122px;">1000000</td> 
 <td style="width:392px;">客户端保持的最大发送请求数。 该配置应用于每一个 broker 连接.&nbsp; Type: integer</td> 
 </tr> 
 <tr> 
 <td>metadata.request.timeout.ms</td> 
 <td>*</td> 
 <td>10 .. 900000</td> 
 <td style="width:122px;">60000</td> 
 <td style="width:392px;">无数据请求超时时间，毫秒。 适用于 metadata 请求等。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>topic.metadata.refresh.interval.ms</td> 
 <td>*</td> 
 <td>-1 .. 3600000</td> 
 <td style="width:122px;">300000</td> 
 <td style="width:392px;">Topic metadata 刷新间隔，毫秒。metadata 自动刷新错误和连接。设置为 -1 关闭刷新间隔。&nbsp; Type: integer</td> 
 </tr> 
 <tr> 
 <td>metadata.max.age.ms</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">参考&nbsp;<code>topic.metadata.refresh.interval.ms</code></td> 
 </tr> 
 <tr> 
 <td>topic.metadata.refresh.fast.cnt</td> 
 <td>*</td> 
 <td>0 .. 1000</td> 
 <td style="width:122px;">10</td> 
 <td style="width:392px;">当 topic 丢失 leader， metadata 请求的发送次数，发送间隔是&nbsp;<code>topic.metadata.refresh.fast.interval.ms</code>而不是&nbsp;<code>topic.metadata.refresh.interval.ms</code>。 该配置用于快速修复broker leader。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>topic.metadata.refresh.fast.interval.ms</td> 
 <td>*</td> 
 <td>1 .. 60000</td> 
 <td style="width:122px;">250</td> 
 <td style="width:392px;">参考&nbsp;<code>topic.metadata.refresh.fast.cnt</code>。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>topic.metadata.refresh.sparse</td> 
 <td>*</td> 
 <td>true, false</td> 
 <td style="width:122px;">true</td> 
 <td style="width:392px;">极少的 metadata 请求 (消费者的网络带宽很小)&nbsp; Type: boolean</td> 
 </tr> 
 <tr> 
 <td>topic.blacklist</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">Topic 黑名单，逗号分隔的正则表达式列表，匹配topic名字，匹配到的 topic 如果不存在，就在 broker metadata 信息中忽略。 Type: pattern list</td> 
 </tr> 
 <tr> 
 <td>debug</td> 
 <td>*</td> 
 <td>generic, broker, topic, metadata, queue, msg, protocol, cgrp, security, fetch, feature, all</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">逗号分隔的列表，控制 debug 上下文。调试生产者：broker,topic,msg。调试消费者：cgrp,topic,fetch&nbsp; Type: CSV flags</td> 
 </tr> 
 <tr> 
 <td>socket.timeout.ms</td> 
 <td>*</td> 
 <td>10 .. 300000</td> 
 <td style="width:122px;">60000</td> 
 <td style="width:392px;">网络请求超时时间。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>socket.blocking.max.ms</td> 
 <td>*</td> 
 <td>1 .. 60000</td> 
 <td style="width:122px;">100</td> 
 <td style="width:392px;">broker 在 socket 操作时最大阻塞时间。值越低，响应越快，但会略微提高CPU使用率。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>socket.send.buffer.bytes</td> 
 <td>*</td> 
 <td>0 .. 100000000</td> 
 <td style="width:122px;">0</td> 
 <td style="width:392px;">Broker socket 发送缓冲大小。系统默认为 0。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>socket.receive.buffer.bytes</td> 
 <td>*</td> 
 <td>0 .. 100000000</td> 
 <td style="width:122px;">0</td> 
 <td style="width:392px;">Broker socket 接收缓冲大小。系统默认为 0。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>socket.keepalive.enable</td> 
 <td>*</td> 
 <td>true, false</td> 
 <td style="width:122px;">false</td> 
 <td style="width:392px;">Broker sockets 允许 TCP 保持活力 (SO_KEEPALIVE)。 Type: boolean</td> 
 </tr> 
 <tr> 
 <td>socket.max.fails</td> 
 <td>*</td> 
 <td>0 .. 1000000</td> 
 <td style="width:122px;">3</td> 
 <td style="width:392px;">Broker 关闭连接的最大错误次数(e.g., timed out requests)。0不关闭。提示：连接自动重新建立。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>broker.address.ttl</td> 
 <td>*</td> 
 <td>0 .. 86400000</td> 
 <td style="width:122px;">1000</td> 
 <td style="width:392px;">保存 broker 地址响应结果的时间 (毫秒)。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>broker.address.family</td> 
 <td>*</td> 
 <td>any, v4, v6</td> 
 <td style="width:122px;">any</td> 
 <td style="width:392px;">允许的 broker IP 地址族：any, v4, v6。 Type: enum value</td> 
 </tr> 
 <tr> 
 <td>reconnect.backoff.jitter.ms</td> 
 <td>*</td> 
 <td>0 .. 3600000</td> 
 <td style="width:122px;">500</td> 
 <td style="width:392px;">通过这个值调节 broker 重连尝试 +-50%。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>statistics.interval.ms</td> 
 <td>*</td> 
 <td>0 .. 86400000</td> 
 <td style="width:122px;">0</td> 
 <td style="width:392px;"> librdkafka 统计间隔。 应用程序需要通过&nbsp;<code>rd_kafka_conf_set_stats_cb()</code>设置统计的回调函数。粒度是 1000ms. 0 关闭统计。 Type: integer </td> 
 </tr> 
 <tr> 
 <td>enabled_events</td> 
 <td>*</td> 
 <td>0 .. 2147483647</td> 
 <td style="width:122px;">0</td> 
 <td style="width:392px;">参考&nbsp;<code>rd_kafka_conf_set_events()</code>。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>error_cb</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">错误回调函数 (参考 rd_kafka_conf_set_error_cb())&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>throttle_cb</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">调节回调函数 (参考 rd_kafka_conf_set_throttle_cb())&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>stats_cb</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">统计回调函数 (参考 rd_kafka_conf_set_stats_cb())&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>log_cb</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">日志回调函数 (参考 rd_kafka_conf_set_log_cb())&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>log_level</td> 
 <td>*</td> 
 <td>0 .. 7</td> 
 <td style="width:122px;">6</td> 
 <td style="width:392px;">日志界别 (syslog(3) levels)&nbsp; Type: integer</td> 
 </tr> 
 <tr> 
 <td>log.thread.name</td> 
 <td>*</td> 
 <td>true, false</td> 
 <td style="width:122px;">false</td> 
 <td style="width:392px;">在日志消息中打印内部线程名。(useful for debugging librdkafka internals)&nbsp; Type: boolean</td> 
 </tr> 
 <tr> 
 <td>log.connection.close</td> 
 <td>*</td> 
 <td>true, false</td> 
 <td style="width:122px;">true</td> 
 <td style="width:392px;">记录 broker 断开连接。由于受 0.9 版本 broker 的&nbsp;<code>connection.max.idle.ms</code>&nbsp;的影响，最好关闭。 Type: boolean</td> 
 </tr> 
 <tr> 
 <td>socket_cb</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">为Socket创建回调函数提供无缝 CLOEXEC&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>open_cb</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">为文件打开回调函数提供无缝 CLOEXEC&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>opaque</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">对应用程序不开放 (set with rd_kafka_conf_set_opaque())&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>default_topic_conf</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">默认 topic 配置，用于自动订阅 topics&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>internal.termination.signal</td> 
 <td>*</td> 
 <td>0 .. 128</td> 
 <td style="width:122px;">0</td> 
 <td style="width:392px;">用于 librdkafka 调用 rd_kafka_destroy() 快速终止的信号。如果没有设置信号， 终止过程会延迟直到所有内部线程的系统调用超时返回，且 rd_kafka_wait_destroyed() 返回 true。如果设置了信号，延迟会最小化。应用程序需要屏蔽该信号，而作为内部信号句柄。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>api.version.request</td> 
 <td>*</td> 
 <td>true, false</td> 
 <td style="width:122px;">false</td> 
 <td style="width:392px;">请求 broker 支持的API版本，调整可用协议特性的功能。如果设置为false，将使用&nbsp;<code>broker.version.fallback</code>设置的回退版本。&nbsp;提示: 依赖的 broker 版本 &gt;=0.10.0。如果 broker（老版本）不支持该请求，使用&nbsp;<code>broker.version.fallback</code>设置的回退版本。 Type: boolean</td> 
 </tr> 
 <tr> 
 <td>api.version.fallback.ms</td> 
 <td>*</td> 
 <td>0 .. 604800000</td> 
 <td style="width:122px;">1200000</td> 
 <td style="width:392px;">配置 ApiVersionRequest 失败多长时间后，使用&nbsp;<code>broker.version.fallback</code>&nbsp;回退版本。提示: ApiVersionRequest 只用新的 broker 能使用。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>broker.version.fallback</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">0.9.0</td> 
 <td style="width:392px;">老版本的 broker（&lt;0.10.0）不支持客户端查询<code>支持协议特性</code>(ApiVersionRequest, see&nbsp;<code>api.version.request</code>)，所以要客户端不知道什么特性可以使用。 用户使用本属性指示 broker 版本，如果 ApiVersionRequest 失败（或不可用），客户端据此属性自动调整特性。与<code>api.version.fallback.ms</code>&nbsp;配合使用。有效值：0.9.0, 0.8.2, 0.8.1, 0.8.0.&nbsp; Type: string</td> 
 </tr> 
 <tr> 
 <td>security.protocol</td> 
 <td>*</td> 
 <td>plaintext, ssl, sasl_plaintext, sasl_ssl</td> 
 <td style="width:122px;">plaintext</td> 
 <td style="width:392px;">与 broker 通讯的协议。 Type: enum value</td> 
 </tr> 
 <tr> 
 <td>ssl.cipher.suites</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">密码套件是个组合体，包括鉴权，加密，认证和秘钥交换程序，用于网络连接的安全设置交换，使用 TLS 或 SSL 网络协议。查看手册&nbsp;<code>ciphers(1)</code>&nbsp;和 `SSL_CTX_set_cipher_list(3)。&nbsp; Type: string</td> 
 </tr> 
 <tr> 
 <td>ssl.key.location</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">客户端的私钥(PEM)路径，用于鉴权。 Type: string</td> 
 </tr> 
 <tr> 
 <td>ssl.key.password</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">私钥密码。 Type: string</td> 
 </tr> 
 <tr> 
 <td>ssl.certificate.location</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">客户端的公钥(PEM)路径，用于鉴权。 Type: string</td> 
 </tr> 
 <tr> 
 <td>ssl.ca.location</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">CA 证书文件或路径，用于校验 broker key。 Type: string</td> 
 </tr> 
 <tr> 
 <td>ssl.crl.location</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">CRL 路径，用于 broker 的证书校验。 Type: string</td> 
 </tr> 
 <tr> 
 <td>sasl.mechanisms</td> 
 <td>*</td> 
 <td>GSSAPI, PLAIN</td> 
 <td style="width:122px;">GSSAPI</td> 
 <td style="width:392px;">使用 SASL 机制鉴权。 支持：GSSAPI, PLAIN.&nbsp;提示: 只能配置一种机制名。 Type: string</td> 
 </tr> 
 <tr> 
 <td>sasl.kerberos.service.name</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">kafka</td> 
 <td style="width:392px;">Kafka 运行的 Kerberos 首要名。 Type: string</td> 
 </tr> 
 <tr> 
 <td>sasl.kerberos.principal</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">kafkaclient</td> 
 <td style="width:392px;">客户端的 Kerberos 首要名。 Type: string</td> 
 </tr> 
 <tr> 
 <td>sasl.kerberos.kinit.cmd</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;"> kinit -S “%{sasl.kerberos. service.name}/%{broker.name}” -k -t “%{sasl.kerberos. keytab}” %{sasl.kerberos. principal} </td> 
 <td style="width:392px;">完整的 kerberos kinit 命令串，%{config.prop.name} 替换为与配置对象一直的值，%{broker.name} broker 的主机名。 Type: string</td> 
 </tr> 
 <tr> 
 <td>sasl.kerberos.keytab</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">Kerberos keytab 文件的路径。如果不设置，则使用系统默认的。提示：不会自动使用，必须在&nbsp;<code>sasl.kerberos.kinit.cmd</code>&nbsp;中添加到模板，如&nbsp;<code>... -t %{sasl.kerberos.keytab}</code>。&nbsp; Type: string</td> 
 </tr> 
 <tr> 
 <td>sasl.kerberos.min.time.before.relogin</td> 
 <td>*</td> 
 <td>1 .. 86400000</td> 
 <td style="width:122px;">60000</td> 
 <td style="width:392px;">Key 恢复尝试的最小时间，毫秒。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>sasl.username</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">使用 PLAIN 机制时，SASL 用户名。 Type: string</td> 
 </tr> 
 <tr> 
 <td>sasl.password</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">使用 PLAIN 机制时，SASL 密码。 Type: string</td> 
 </tr> 
 <tr> 
 <td>group.id</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">客户端分组字符串。同组的客户端使用相同的 group.id。 Type: string</td> 
 </tr> 
 <tr> 
 <td>partition.assignment.strategy</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">range,roundrobin</td> 
 <td style="width:392px;">partition 分配策略，当选举组 leader 时，分配 partition 给组成员的策略。 Type: string</td> 
 </tr> 
 <tr> 
 <td>session.timeout.ms</td> 
 <td>*</td> 
 <td>1 .. 3600000</td> 
 <td style="width:122px;">30000</td> 
 <td style="width:392px;">客户端组会话探测失败超市时间。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>heartbeat.interval.ms</td> 
 <td>*</td> 
 <td>1 .. 3600000</td> 
 <td style="width:122px;">1000</td> 
 <td style="width:392px;">组会话保活心跳间隔。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>group.protocol.type</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">consumer</td> 
 <td style="width:392px;">组协议类型。 Type: string</td> 
 </tr> 
 <tr> 
 <td>coordinator.query.interval.ms</td> 
 <td>*</td> 
 <td>1 .. 3600000</td> 
 <td style="width:122px;">600000</td> 
 <td style="width:392px;">多久查询一次当前的客户端组协调人。如果当前的分配协调人挂了，为了更快的恢复协调人，探测时间间隔会除以 10。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>enable.auto.commit</td> 
 <td>C</td> 
 <td>true, false</td> 
 <td style="width:122px;">true</td> 
 <td style="width:392px;">在后台周期性的自动提交偏移量。 Type: boolean</td> 
 </tr> 
 <tr> 
 <td>auto.commit.interval.ms</td> 
 <td>C</td> 
 <td>0 .. 86400000</td> 
 <td style="width:122px;">5000</td> 
 <td style="width:392px;">消费者偏移量提交（写入）到存储的频率，毫秒。(0 = 不可用)&nbsp; Type: integer</td> 
 </tr> 
 <tr> 
 <td>enable.auto.offset.store</td> 
 <td>C</td> 
 <td>true, false</td> 
 <td style="width:122px;">true</td> 
 <td style="width:392px;">为应用程序自动保存最后消息的偏移量。 Type: boolean</td> 
 </tr> 
 <tr> 
 <td>queued.min.messages</td> 
 <td>C</td> 
 <td>1 .. 10000000</td> 
 <td style="width:122px;">100000</td> 
 <td style="width:392px;">每一个 topic+partition，本地消费者队列的最小消息数。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>queued.max.messages.kbytes</td> 
 <td>C</td> 
 <td>1 .. 1000000000</td> 
 <td style="width:122px;">1000000</td> 
 <td style="width:392px;">每一个 topic+partition，本地消费者队列的最大大小，单位kilobytes。该值应该大于 fetch.message.max.bytes。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>fetch.wait.max.ms</td> 
 <td>C</td> 
 <td>0 .. 300000</td> 
 <td style="width:122px;">100</td> 
 <td style="width:392px;">为写满<code>fetch.min.bytes</code>，broker 的最大等待时间。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>fetch.message.max.bytes</td> 
 <td>C</td> 
 <td>1 .. 1000000000</td> 
 <td style="width:122px;">1048576</td> 
 <td style="width:392px;">每一个 topic+partition 初始化的最大大小（bytes）用于从 broker 读消息。如果客户端遇到消息大于这个值，会逐步扩大直到塞下这个消息。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>max.partition.fetch.bytes</td> 
 <td>C</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">参考&nbsp;<code>fetch.message.max.bytes</code></td> 
 </tr> 
 <tr> 
 <td>fetch.min.bytes</td> 
 <td>C</td> 
 <td>1 .. 100000000</td> 
 <td style="width:122px;">1</td> 
 <td style="width:392px;">broker 请求的最小数据大小，单位bytes。如果达到 fetch.wait.max.ms 时间，则不管这个配置，将已收到的数据发送给客户端。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>fetch.error.backoff.ms</td> 
 <td>C</td> 
 <td>0 .. 300000</td> 
 <td style="width:122px;">500</td> 
 <td style="width:392px;">对于 topic+partition，如果接受错误，下一个接受请求间隔多长时间。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>offset.store.method</td> 
 <td>C</td> 
 <td>none, file, broker</td> 
 <td style="width:122px;">broker</td> 
 <td style="width:392px;">偏移量存储方式：’file’ - 本地文件存储 (offset.store.path, et.al), ‘broker’ - 在 broker 上提交存储 (要求 Apache Kafka 0.8.2 或以后版本)。 Type: enum value</td> 
 </tr> 
 <tr> 
 <td>consume_cb</td> 
 <td>C</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">消息消费回调函数 (参考 rd_kafka_conf_set_consume_cb())&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>rebalance_cb</td> 
 <td>C</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">消费者组重新分配后调用 (参考 rd_kafka_conf_set_rebalance_cb())&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>offset_commit_cb</td> 
 <td>C</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">偏移量提交结果回调函数 (参考 rd_kafka_conf_set_offset_commit_cb())&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>enable.partition.eof</td> 
 <td>C</td> 
 <td>true, false</td> 
 <td style="width:122px;">true</td> 
 <td style="width:392px;">当消费者到达分区结尾，发送 RD_KAFKA_RESP_ERR__PARTITION_EOF 事件。 Type: boolean</td> 
 </tr> 
 <tr> 
 <td>queue.buffering.max.messages</td> 
 <td>P</td> 
 <td>1 .. 10000000</td> 
 <td style="width:122px;">100000</td> 
 <td style="width:392px;">生产者队列允许的最大消息数。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>queue.buffering.max.kbytes</td> 
 <td>P</td> 
 <td>1 .. 2147483647</td> 
 <td style="width:122px;">4000000</td> 
 <td style="width:392px;">生产者队列允许的最大大小，单位kb。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>queue.buffering.max.ms</td> 
 <td>P</td> 
 <td>1 .. 900000</td> 
 <td style="width:122px;">1000</td> 
 <td style="width:392px;">生产者队列缓存数据的最大时间，毫秒。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>message.send.max.retries</td> 
 <td>P</td> 
 <td>0 .. 10000000</td> 
 <td style="width:122px;">2</td> 
 <td style="width:392px;">消息集发送失败重试次数。提示&nbsp;重试会导致重排。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>retries</td> 
 <td>P</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">参考&nbsp;<code>message.send.max.retries</code></td> 
 </tr> 
 <tr> 
 <td>retry.backoff.ms</td> 
 <td>P</td> 
 <td>1 .. 300000</td> 
 <td style="width:122px;">100</td> 
 <td style="width:392px;">重试消息发送前的补偿时间。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>compression.codec</td> 
 <td>P</td> 
 <td>none, gzip, snappy, lz4</td> 
 <td style="width:122px;">none</td> 
 <td style="width:392px;">压缩消息集使用的压缩编解码器。这里配置的是所有 topic 的默认值，可能会被 topic 上的&nbsp;<code>compression.codec</code>&nbsp;属性覆盖。 Type: enum value</td> 
 </tr> 
 <tr> 
 <td>batch.num.messages</td> 
 <td>P</td> 
 <td>1 .. 1000000</td> 
 <td style="width:122px;">10000</td> 
 <td style="width:392px;">一个消息集最大打包消息数量。整个消息集的大小仍受限于&nbsp;<code>message.max.bytes</code>。&nbsp; Type: integer</td> 
 </tr> 
 <tr> 
 <td>delivery.report.only.error</td> 
 <td>P</td> 
 <td>true, false</td> 
 <td style="width:122px;">false</td> 
 <td style="width:392px;">只对失败的消息提供分发报告。 Type: boolean</td> 
 </tr> 
 <tr> 
 <td>dr_cb</td> 
 <td>P</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">分发报告回调函数 (参考 rd_kafka_conf_set_dr_cb())&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>dr_msg_cb</td> 
 <td>P</td> 
 <td>&nbsp;</td> 
 <td style="width:122px;">&nbsp;</td> 
 <td style="width:392px;">分发报告回调函数 (参考 rd_kafka_conf_set_dr_msg_cb())&nbsp; Type: pointer</td> 
 </tr> 
 </tbody> 
</table>

## Topic 配置属性 ##

<table> 
 <thead> 
 <tr> 
 <th>属性</th> 
 <th>C/P</th> 
 <th>范围</th> 
 <th>默认值</th> 
 <th>描述</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>request.required.acks</td> 
 <td>P</td> 
 <td>-1 .. 1000</td> 
 <td>1</td> 
 <td>这个字段标示 leader broker 要从 ISR broker 接收多少个 ack，然后才确认发送请求：0=不发送任何 response/ack 给客户端,&nbsp;1=只有 leader broker 需要 ack 消息,&nbsp;-1&nbsp;or&nbsp;all=broker 阻塞直到所有的同步备份(ISRs)或<code>in.sync.replicas</code>设置的备份返回消息提交的确认应答。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>acks</td> 
 <td>P</td> 
 <td>&nbsp;</td> 
 <td>&nbsp;</td> 
 <td>参考&nbsp;<code>request.required.acks</code></td> 
 </tr> 
 <tr> 
 <td>request.timeout.ms</td> 
 <td>P</td> 
 <td>1 .. 900000</td> 
 <td>5000</td> 
 <td>生产者请求等待应答的超时时间，毫秒。这个值仅在 broker 上强制执行。参考&nbsp;<code>request.required.acks</code>，不能等于 0.&nbsp; Type: integer</td> 
 </tr> 
 <tr> 
 <td>message.timeout.ms</td> 
 <td>P</td> 
 <td>0 .. 900000</td> 
 <td>300000</td> 
 <td>本地消息超时时间。这个值仅在本地强制执行，限制生产的消息等待被成功发送的等待时间，0 是不限制。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>produce.offset.report</td> 
 <td>P</td> 
 <td>true, false</td> 
 <td>false</td> 
 <td>报告生产消息的偏移量给应用程序。应用程序必须使用&nbsp;<code>dr_msg_cb</code>&nbsp;从&nbsp;<code>rd_kafka_message_t.offset</code>&nbsp;中获取偏移量。 Type: boolean</td> 
 </tr> 
 <tr> 
 <td>partitioner_cb</td> 
 <td>P</td> 
 <td>&nbsp;</td> 
 <td>&nbsp;</td> 
 <td>分区方法回调函数 (参考 rd_kafka_topic_conf_set_partitioner_cb())&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>opaque</td> 
 <td>*</td> 
 <td>&nbsp;</td> 
 <td>&nbsp;</td> 
 <td>应用程序不可见 (参考 rd_kafka_topic_conf_set_opaque())&nbsp; Type: pointer</td> 
 </tr> 
 <tr> 
 <td>compression.codec</td> 
 <td>P</td> 
 <td>none, gzip, snappy, lz4, inherit</td> 
 <td>inherit</td> 
 <td>压缩消息集的压缩编解码器。 Type: enum value</td> 
 </tr> 
 <tr> 
 <td>auto.commit.enable</td> 
 <td>C</td> 
 <td>true, false</td> 
 <td>true</td> 
 <td> 如果是 true，周期性的提交最后一个消息的偏移量。用于当程序重启时抛弃不用的消息。 如果是 false，应用程序需要调用&nbsp;<code>rd_kafka_offset_store()</code>&nbsp;保存偏移量 (可选). 提示&nbsp;这个属性时能用于简单消费者，high-level KafkaConsumer 会被全局的&nbsp;<code>enable.auto.commit</code>&nbsp;属性替代。 提示&nbsp;目前没有整合 zookeeper，根据&nbsp;<code>offset.store.method</code>&nbsp;的配置 偏移量将写入 broker 或本地文件 file according to offset.store.method.&nbsp; Type: boolean </td> 
 </tr> 
 <tr> 
 <td>enable.auto.commit</td> 
 <td>C</td> 
 <td>&nbsp;</td> 
 <td>&nbsp;</td> 
 <td>参考&nbsp;<code>auto.commit.enable</code></td> 
 </tr> 
 <tr> 
 <td>auto.commit.interval.ms</td> 
 <td>C</td> 
 <td>10 .. 86400000</td> 
 <td>60000</td> 
 <td>消费者偏移量提交（写入）到存储的频率，毫秒。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>auto.offset.reset</td> 
 <td>C</td> 
 <td>smallest, earliest, beginning, largest, latest, end, error</td> 
 <td>largest</td> 
 <td> 如果偏移量存储还没有初始化或偏移量超过范围时的处理方式：Action to take when there is no initial offset in offset store or the desired offset is out of range: ‘smallest’,’earliest’ - 自动重设偏移量为最小偏移量， ’largest’,’latest’ - 自动重设偏移量为最大偏移量， ’error’ - 通过消费消息触发一个错误，请检查&nbsp;<code>message-&gt;err</code>。&nbsp; Type: enum value </td> 
 </tr> 
 <tr> 
 <td>offset.store.path</td> 
 <td>C</td> 
 <td>&nbsp;</td> 
 <td>.</td> 
 <td>存储偏移量的本地文件路径。如果路径是个目录，在目录下自动创建基于 topic 和 partition 的文件名。 Type: string</td> 
 </tr> 
 <tr> 
 <td>offset.store.sync.interval.ms</td> 
 <td>C</td> 
 <td>-1 .. 86400000</td> 
 <td>-1</td> 
 <td>偏移量文件 fsync() 的间隔，毫秒。-1 不同步，0 每次写入后立即同步。 Type: integer</td> 
 </tr> 
 <tr> 
 <td>offset.store.method</td> 
 <td>C</td> 
 <td>file, broker</td> 
 <td>broker</td> 
 <td>偏移量存储方式：’file’ - 本地文件存储 (offset.store.path, et.al), ‘broker’ - 在 broker 上提交存储 (要求 Apache Kafka 0.8.2 或以后版本)。 Type: enum value</td> 
 </tr> 
 <tr> 
 <td>consume.callback.max.messages</td> 
 <td>C</td> 
 <td>0 .. 1000000</td> 
 <td>0</td> 
 <td>一次&nbsp;<code>rd_kafka_consume_callback*()</code>&nbsp;调配的最大消息数 (0 = 无限制)&nbsp; Type: integer</td> 
 </tr> 
 </tbody> 
</table>

### C/P 含义：C = 生产者, P = 消费者, \* = 二者都有 ###