首页 > 代码库 > kafka0.9.0及0.10.0配置属性

kafka0.9.0及0.10.0配置属性

2024-08-29 23:02:45   222人阅读

问题导读

1.borker包含哪些属性？
2.Producer包含哪些属性？
3.Consumer如何配置？

技术分享




borker(0.9.0及0.10.0)配置
Kafka日志本身是由多个日志段组成(log segment)。一个日志是一个FileMessageSet，它包含了日志数据以及OffsetIndex对象，该对象使用位移来读取日志数据 
* borker配置就是指配置server.properties文件 *


最小配置

通常情况下需要在减压缩kafka后，修改config/server.properties 配置文件中的以下两项

[Bash shell] 纯文本查看 复制代码
?
1
2
3
log.dirs = kafka-logs
zookeeper.connect = localhost:9092
listeners = PLAINTEXT://ip:9092


  • log.dirs 指的是kafka的log Data保存的目录，默认为Null。如果不指定log Data会保存到log.dir设置的目录中，log.dir默认为/tmp/kafka-logs。需要保证启动KafKaServer的用户对log.dirs或log.dir设置的目录具有读与写的权限。
  • zookeeper.connect 指的是zookeeper集群的地址，可以是多个，多个之间用逗号分割hostname1:port1,hostname2:port2,hostname3:port3
  • listeners 监听列表(以逗号分隔 不同的协议(如plaintext,trace,ssl、不同的IP和端口))


kafka提供的borker配置


[Bash shell] 纯文本查看 复制代码
?
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# Replication configurations
num.replica.fetchers=4
replica.fetch.max.bytes=1048576
replica.fetch.wait.max.ms=500
replica.high.watermark.checkpoint.interval.ms=5000
replica.socket.timeout.ms=30000
replica.socket.receive.buffer.bytes=65536
replica.lag.time.max.ms=10000
 
controller.socket.timeout.ms=30000
controller.message.queue.size=10
 
# Log configuration
num.partitions=8
message.max.bytes=1000000
auto.create.topics.enable=true
log.index.interval.bytes=4096
log.index.size.max.bytes=10485760
log.retention.hours=168
log.flush.interval.ms=10000
log.flush.interval.messages=20000
log.flush.scheduler.interval.ms=2000
log.roll.hours=168
log.retention.check.interval.ms=300000
log.segment.bytes=1073741824
 
# ZK configuration
zookeeper.connection.timeout.ms=6000
zookeeper.sync.time.ms=2000
 
# Socket server configuration
num.io.threads=8
num.network.threads=8
socket.request.max.bytes=104857600
socket.receive.buffer.bytes=1048576
socket.send.buffer.bytes=1048576
queued.max.requests=16
fetch.purgatory.purge.interval.requests=100
producer.purgatory.purge.interval.requests=100


配置详细说明


名称 说明 类型 默认值 有效值 重要性
zookeeper.connect zookeeper集群的地址，
可以是多个，
多个之间用逗号分割		 string localhost:
2181		 ip1
:port1
,ip2
:port2		 高
zookeeper.connection.timeout.ms 客户端在建立通zookeeper
连接中的最大等待时间		 int null 6000 高
zookeeper.session.timeout.ms ZooKeeper的session的超时
时间，如果在这段时间内没
有收到ZK的心跳，则会被认
为该Kafka server挂掉了。
如果把这个值设置得过低可
能被误认为挂掉，如果设置
得过高，如果真的挂了，则需
要很长时间才能被server得知		 int 6000   高
zookeeper.sync.time.ms 一个ZK follower能落后leader
的时间		 int 2000   高
listeners 监听列表(以逗号分隔 不同的
协议(如plaintext,trace,ssl、
不同的IP和端口)),hostname
如果设置为0.0.0.0则绑定所
有的网卡地址；如果hostname
为空则绑定默认的网卡。
如果
没有配置则默认为
java.net
.InetAddress
.getCanonicalHostName()		 string null 如：PLAINTEXT:
//myhost:
9092,
TRACE://:
9091
或 PLAINTEXT:
//0.0.0.0
:9092,		 高
host.name 。如果设置了它，
会仅绑定这个
地址。
如果没有设置，则会
绑定所有
的网络接口，并提交
一个给ZK。
不推荐使用 只有当
listeners没
有设置时才有必要使用。		 string “’ 如：
”localhost”		 高
port server用来接受
client连接的端口。
不推荐使用,使用
listeners配置
项代替；只有在
listeners没有
配置时才使用。		 int 9092   高
advertised.host.name 会将hostname通知给
生产者
和消费者，在多网卡
时需要
设置该值为另一个ip地址。
如果没有设置该值，
则返回 
配置项host.name设置的值，
如果host.name没有设
置则返回java.net.InetAddress.
getCanonicalHostName()
不推荐使用 只有当
advertised.listeners或
listeners没有设置时才
有必要使用。		 string null   高
advertised.listeners 设置不同于listeners配置
的监听列表即不同于
listeners设置的网卡地址
及端口；如果没有配置，
会使用listeners的值		 string null   高
advertised.port 分发这个端口给所有的
producer，consumer和
其他broker来建立连接
。如果此端口跟server
绑定的端口不同，
则才有必要设置。
不推荐使用 只有当
advertised.listeners
或listeners没有设置
时才有必要使用。		 int null   高
auto.create.topics.enable 是否允许自动创建topic。
如果设为true，那么
produce，consume
或者fetch metadata
一个不存在的topic时，
就会自动创建一个默认
replication factor和
partition number的topic。		 boolean true   高
background.threads 一些后台任务处理的
线程数，例如过期消
息文件的删除等，
一般情况下不需要去
做修改		 int 10   高
broker.id 每一个broker在集群中
的唯一表示，要求是正数。
当该服务器的IP地址发
生改变时，broker.id没有
变化，则不会影响
consumers的消息情况。		 int -1   高
compression.type 指定topic的压缩类型。
除了支持’gzip’, ‘snappy’,
‘lz4’外，还支持
”uncompressed(不压缩)
”以及produce
r(由producer来指定)		 string producer   高
delete.topic.enable 是否启动删除topic。
如果设置为false,
你在删除topic的时
候无法删除，但是会打
上一个你将删除该topic
的标记，等到你修改这
一属性的值为true后重
新启动Kafka集群的时候
，集群自动将那些标记
删除的topic删除掉，对应
的log.dirs目录下的topic
目录和数据也会被删除。
而将这一属性设置为true之后，
你就能成功删除你想要
删除的topic了		 boolean false   高
auto.leader.rebalance.enable 一个后台线程会周期性
的自动尝试，为所有的
broker的每个partition
平衡leadership，使kafka
的leader均衡。		 boolean true   高
leader.imbalance.check.interval.seconds 检查leader是否均衡的
时间间隔(秒)		 long 300   高
leader.imbalance.per.broker.percentage 每个broker允许的不平衡
的leader的百分比。
如果每个broker超过
了这个百分比，复制控
制器会重新平衡leadership。		 int 10   高
log.flush.interval.messages 数据flush(sync)到硬盘
前之前累积的消息条数
，因为磁盘IO操作是一
个慢操作,但又是一个
”数据可靠性”的必要
手段,所以此参数的设置
,需要在”数据可靠性”
与”性能”之间做必要
的权衡.如果此值过大,
将会导致每次”fsync”
的时间较长
(IO阻塞),如果此值过
小,将会导致”fsync”的
次数较多,这也意味着
整体的client请求有一
定的延迟.物理server
故障,将会导致没有
fsync的消息丢失		 long 9223372
0368547
75807
（此为一
个数字）		   高
log.flush.interval.ms 当达到下面的时间
(ms)时，执行一次
强制的flush操作。
interval.ms和
interval.messages
无论哪个达到，
都会flush。		 long null   高
log.flush.offset.checkpoint.interval.ms 记录上次把log刷
到磁盘的时间点的
频率，用来日后的
recovery。通常
不需要改变		 long 60000   高
log.flush.scheduler.interval.ms 检查是否需要固
化到硬盘的时间
间隔		 long 92233720
3685477
5807
（此为一
个数字）		   高
log.retention.bytes topic每个分区的
最大文件大小，
一个topic的大小
限制 = 分区数*
log.retention.bytes。
-1没有大小限
log.retention.bytes和log.retention.minutes
任意一个达到要求，
都会执行删除，
会被topic创建时
的指定参数覆盖		 loong -1   高
log.retention.hours 日志保存时间，
默认为7天（168小时）
。超过这个时间会根
据policy处理数据。
bytes和minutes无论
哪个先达到都会触发		 int 168   高
log.retention.minutes 数据存储的最大时间
超过这个时间会根据
log.cleanup.policy
设置的策略处理数
据，也就是消费端能
够多久去消费数据		        
log.retention.bytes和log.retention.minutes任意一个达到要求，都会执行删除，会被topic创建时的指定参数覆盖 int null   高  
log.roll.hous 当达到下面时间，
会强制新建一个segment。
这个参数会在日志
segment没有达到
log.segment.bytes
设置的大小，也会强制
新建一个segment会		 int 168   高
log.roll.jitter.{ms,hours} 从logRollTimeMillis抽
离的jitter最大数目		 int 0   高
log.segment.bytes topic partition的日志存
放在某个目录下诸多
文件中，这些文件将
partition的日志切分成
一段一段的；这个属性就
是每个文件的最大尺寸；
当尺寸达到这个数值时，
就会创建新文件。
此设置可以由每个topic
基础设置时进行覆盖		 long 1G=
1024*
1024*
1024		   高
log.segment.delet.delay.ms 删除文件系统上文件
的等待时间，默认是
1分钟		 long 6000   高
message.max.bytes 表示一个服务器能够
接收处理的消息的最
大字节数，注意这个
值producer和consumer
必须设置一致，且不要大于fetch.message.max.bytes
属性的值该值默认是
1000012字节，大概900KB		        
int 1000012   高    
min.insync.replicas 该属性规定了最小的
ISR数。当producer设置request.required.acks
为all或-1时，指定副
本(replicas)的最小数目
（必须确认每一个repica
的写数据都是成功的），
如果这个数目没有达到，
producer会产生异常。		 int 1   高
num.io.threads 服务器用来处理请求
的I/O线程的数目；
这个线程数目至少要
等于硬盘的个数。		 int 8   高
num.network.threads 服务器用来处理网络
请求的网络线程数目；
一般你不需要更改这
个属性		 int 3   高
num.recovery.threads.per.data.dir 每数据目录用于日志
恢复启动和关闭冲洗
时的线程数量		 int 1   高
num.replica.fetchers 从leader进行复制
消息的线程数，增大这个
数值会增加follower的IO		 int 1   高
offset.metadata.max.bytes 允许client(消费者)保存
它们元数据(offset)的
最大的数据量		 int 4096(4kb)    
offsets.commit.required.acks 在offset commit可以接
受之前，需要设置确认的
数目，一般不需要更改		 int -1   高
offsets.commit.timeout.ms offset commit会延迟
直至此超时或所需的
副本数都收到offset 
commit，
这类似于producer请
求的超时		 int 5000   高
offsets.load.buffer.size 此设置对应于offset
manager在读取缓存
offset segment的批量
大小（以字节为单位).		 int 5242880   高
offsets.retention.check.interval.ms offset管理器检查陈
旧offsets的频率		 long 600000
(10分钟)		   高
offsets.topic.num.partitions 偏移的提交topic的分
区数目。 由于目前不
支持部署之后改变，
我们建议您使用生产
较高的设置
（例如，100-200）		 int 50   高
offsets.topic.replication.factor 复制因子的offset提交topic。
较高的设置
（例如三个或四个），
建议以确保更高的
可用性。如果offset topic
创建时，broker比
复制因子少，
offset topic将以较
少的副本创建。		 short 3   高
offsets.topic.segment.bytes offset topic的Segment
大小。因为它使用
压缩的topic，所有
Sgment的大小应该
保持小一点，以促进更
快的日志压实和负载		 int 104857600   高
queued.max.requests 在网络线程
(network threads)
停止读取新请求之前，
可以排队等待I/O线程
处理的最大请求个数。
若是等待IO的请求超
过这个数值，那么会
停止接受外部消息		 int 500   高
quota.consumer.default 以clientid或
consumer group区分
的consumer端每秒
可以抓取的最大byte		 long 92233720
36854775
807
（此为一
个数字）		   高
quota.producer.default producer端每秒可
以产生的最大byte		 long 92233720
3685477
5807
（此为一
个数字）		   高
replica.fetch.max.bytes replicas每次获取数
据的最大字节数		 int 1048576   高
replica.fetch.min.bytes fetch的最小数据尺寸,
如果leader中尚未同步
的数据不足此值,将会
阻塞,直到满足条件		 int 1   高
replica.fetch.wait.max.ms replicas同leader之间
通信的最大等待时间，
失败了会重试。这个值
须小于
replica.lag.time.max.ms，
以防止低吞吐量
主题ISR频繁收缩		 int 500   高
replica.high.watermark.checkpoint.interval.ms 每一个replica存储自己
的high watermark到磁
盘的频率，用来日后
的recovery		 int 5000   高
replica.socket.timeout.ms 复制数据过程中，
replica发送给leader的
网络请求的socket超
时时间,至少等于
replica.fetch.wait.max.ms		 int 30000   高
replica.socket.receive.buffer.bytes 复制过程leader接
受请求的buffer大小		 int 65536
(64*1024)		   高
replica.lag.time.max.ms replicas响应partition leader
的最长等待时间，
若是超过这个时间，
就将replicas列入
ISR(in-sync replicas)，
并认为它是死的，
不会再加入管理中		 long 10000   高
replica.lag.max.messages 如果follower落后与
leader太多,将会认
为此follower
[或者说partition relicas]
已经失效。 通常,在
follower与leader通讯时,
因为网络延迟或者链
接断开,
总会导致replicas中消息
同步滞后如果消息之
后太多,leader将认为
此follower网络延迟
较大或者消息吞吐
能力有限,将会把此
replicas迁移到其他
follower中.在broker数
量较少,或者网络不足
的环境中,建议提高此值.		 int 4000   高
request.timeout.ms producer等待响应的
最长时间，如果超时
将重发几次，最终报错		 int 30000   高
socket.receive.buffer.bytes socket用于接收网
络请求的缓存大小		 int 102400   高
socket.request.max.bytes server能接受的请求
的最大的大小，
这是为了防止server
跑光内存，不能大
于Java堆的大小。		 int 104857600
(100*1024
*1024)
（此为一
个表达式）		   高
socket.send.buffer.bytes server端用来处理
socket连接的
SO_SNDBUFF缓冲大小		 int 102400   高
controller.socket.timeout.ms partition管理控制
器进行备份时，
socket的超时时间		 int 30000   高
controller.message.queue.size partition leader与
replicas数据同步时,
消息的队列大小		 int 10   高
num.partitions 每个topic的分区个数，
若是在topic创建时候
没有指定的话会被
topic创建时的指定
参数覆盖		 int 1 推荐
设为8		 高
log.index.interval.bytes 当执行一次fetch后，
需要一定的空间扫描
最近的offset，设置的
越大越好，但是也更耗
内存一般使用默认值就可以		 int 4096   中
log.index.size.max.bytes 每个log segment的
最大尺寸。注意，
如果log尺寸达到这
个数值，即使尺寸
没有超过log.segment.bytes
限制，也需要产生新的
log segment。		 int 10485760   中
fetch.purgatory.purge.interval.requests 非立即答复请求放入
purgatory中，
当到达或超出interval
时认为request complete		 int 1000   中
producer.purgatory.purge.interval.requests producer请求清除时间 int 1000   中
default.replication.factor 一个topic ，默认分区
的replication个数 ，
不能大于集群中
broker的个数。		 int 1   中
group.max.session.timeout.ms 注册consumer允许
的最大超时时间		 int 300000   中
group.min.session.timeout.ms 注册consumer允许
的最小超时时间		 int 6000   中
inter.broker.protocol.version broker协议版本 string 0.10.0   中
log.cleaner.backoff.ms 检查log是否需要
clean的时间间隔		 long 15000   中
log.cleaner.dedupe.buffer.size 日志压缩去重时候的
缓存空间，在空间允许
的情况下，越大越好		 long 134217
728

   中
log.cleaner.delete.retention.ms 保存时间；保存压缩
日志的最长时间；
也是客户端消费消息的
最长时间，
同log.retention.minutes
的区别在于一个控制未
压缩数据，一个控制压
缩后的数据；会被topic
创建时的指定时间覆盖。		 long 86400000
(一天)		   中
log.cleaner.enable 是否启动压缩日志,
当这个属性设置为false时
，一旦日志的保存时间或
者大小达到上限时，
就会被删除；
如果设置为true，
则当保存属性达到上限时，
就会进行压缩		 boolean false   中
log.cleaner.threads 日志压缩运行的线程数 int 1   中
log.cleaner.io.buffer.load.factor 日志清理中hash表的
扩大因子，一般不需
要修改		 double 0.9   中
log.cleaner.io.buffer.size log cleaner清除过程
中针对日志进行索引
化以及精简化所用到
的缓存大小。最好设置大
点，以提供充足的内存		 int 524288   中
log.cleaner.io.max.bytes.per.second 进行log compaction时，
log cleaner可以拥有的
最大I/O数目。这项设置
限制了cleaner，以避免干
扰活动的请求服务。		 double 1.79769313
48623157E308
（此为一
个数字）		   中
log.cleaner.min.cleanable.ratio 这项配置控制
log compactor
试图清理日志的频率
（假定[log compaction]
是打开的）。
默认避免清理压缩超过
50%的日志。这个比率
绑定了备份日志所消耗
的最大空间（50%的
日志备份时压缩率为50%）。
更高的比率则意味着浪
费消耗更少，也就可
以更有效的清理更多
的空间。这项设置在
每个topic设置中可以覆盖		 double 0.5   中
log.preallocate 是否预创建新文件，windows推荐使用 boolean false   中
log.retention.check.interval.ms 检查日志分段文件的间隔时间，以确定是否文件属性是否到达删除要求。 long 300000   中
max.connections.per.ip 一个broker允许从每个ip地址连接的最大数目 int 2147483647
=Int.
MaxValue		   中
max.connections.per.ip.overrides 每个IP或主机名覆盖连接的默认最大数量 string “”   中
replica.fetch.backoff.ms 复制数据时失败等待时间 int 1000   中
reserved.broker.max.id broker可以使用的最大ID值 int 1000   中
topic level 配置
broker级别的参数可以由topic级别的覆写，不是所有的broker参数在topic级别都有对应值 
以下是topic-level的配置选项。server的默认配置在Server Default Property列下给出了，设定这些默认值不会改变原有的设置

Property Default Server Default Property Description
cleanup.policy delete log.cleanup.policy 要么是”delete“要么是”compact“； 这个字符串指明了针对旧日志部分的利用方式；默认方式（”delete”）将会丢弃旧的部分当他们的回收时间或者尺寸限制到达时。”compact“将会进行日志压缩
delete.retention.ms 86400000 (24 hours) log.cleaner.delete.retention.ms 对于压缩日志保留的最长时间，也是客户端消费消息的最长时间，通log.retention.minutes的区别在于一个控制未压缩数据，一个控制压缩后的数据。此项配置可以在topic创建时的置顶参数覆盖
flush.messages none log.flush.interval.messages 此项配置指定时间间隔：强制进行fsync日志。例如，如果这个选项设置为1，那么每条消息之后都需要进行fsync，如果设置为5，则每5条消息就需要进行一次fsync。一般来说，建议你不要设置这个值。此参数的设置,需要在”数据可靠性”与”性能”之间做必要的权衡.如果此值过大,将会导致每次”fsync”的时间较长(IO阻塞),如果此值过小,将会导致”fsync”的次数较多,这也意味着整体的client请求有一定的延迟.物理server故障,将会导致没有fsync的消息丢失.
flush.ms None log.flush.interval.ms 此项配置用来置顶强制进行fsync日志到磁盘的时间间隔；例如，如果设置为1000，那么每1000ms就需要进行一次fsync。一般不建议使用这个选项
index.interval.bytes 4096 log.index.interval.bytes 默认设置保证了我们每4096个字节就对消息添加一个索引，更多的索引使得阅读的消息更加靠近，但是索引规模却会由此增大；一般不需要改变这个选项
max.message.bytes 1000000 max.message.bytes kafka追加消息的最大尺寸。注意如果你增大这个尺寸，你也必须增大你consumer的fetch 尺寸，这样consumer才能fetch到这些最大尺寸的消息。
min.cleanable.dirty.ratio 0.5 in.cleanable.dirty.ratio 此项配置控制log压缩器试图进行清除日志的频率。默认情况下，将避免清除压缩率超过50%的日志。这个比率避免了最大的空间浪费
min.insync.replicas 1 min.insync.replicas 当producer设置 acks为-1时，min.insync.replicas指定replicas的最小数目（必须确认每一个repica的写数据都是成功的），如果这个数目没有达到，producer会产生异常。
retention.bytes None log.retention.bytes 如果使用“delete”的retention 策略，这项配置就是指在删除日志之前，日志所能达到的最大尺寸。默认情况下，没有尺寸限制而只有时间限制
retention.ms 7 days log.retention.minutes 如果使用“delete”的retention策略，这项配置就是指删除日志前日志保存的时间。
segment.bytes 1GB log.segment.bytes kafka中log日志是分成一块块存储的，此配置是指log日志划分成块的大小
segment.index.bytes 10MB log.index.size.max.bytes 此配置是有关offsets和文件位置之间映射的索引文件的大小；一般不需要修改这个配置
segment.ms 7 days log.roll.hours 即使log的分块(segment)文件没有达到需要删除、压缩的大小，一旦log 的时间达到这个上限，就会强制新建一个log分块文件
segment.jitter.ms 0 log.roll.jitter.{ms,hours} The maximum jitter to subtract from logRollTimeMillis.


Producer配置

名称 说明 类型 默认值 有效值 重要性
bootstrap.servers 用于建立与kafka集群连接的host/port组。数据将会在所有servers上均衡加载，不管哪些server是指定用于bootstrapping。这个列表仅仅影响初始化的hosts（用于发现全部的servers）。这个列表格式：host1:port1,host2:port2,…因为这些server仅仅是用于初始化的连接，以发现集群所有成员关系（可能会动态的变化），这个列表不需要包含所有的servers（你可能想要不止一个server，尽管这样，可能某个server宕机了）。如果没有server在这个列表出现，则发送数据会一直失败，直到列表可用。 list     高
acks producer需要server接收到数据之后发出的确认接收的信号，此项配置就是指procuder需要多少个这样的确认信号。此配置实际上代表了数据备份的可用性。以下设置为常用选项：（1）acks=0： 设置为0表示producer不需要等待任何确认收到的信息。副本将立即加到socket buffer并认为已经发送。没有任何保障可以保证此种情况下server已经成功接收数据，同时重试配置不会发生作用（因为客户端不知道是否失败）回馈的offset会总是设置为-1；（2）acks=1： 这意味着至少要等待leader已经成功将数据写入本地log，但是并没有等待所有follower是否成功写入。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。（3）acks=all： 这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证。， string -1 [all -1 0 1] 高
key.serializer key的序列化方式，若是没有设置，同serializer.class 实现Serializer接口的class     高
value.serializer value序列化类方式 实现Serializer接口的class     高
buffer.memory producer可以用来缓存数据的内存大小。如果数据产生速度大于向broker发送的速度，producer会阻塞或者抛出异常，以“block.on.buffer.full”来表明。这项设置将和producer能够使用的总内存相关，但并不是一个硬性的限制，因为不是producer使用的所有内存都是用于缓存。一些额外的内存会用于压缩（如果引入压缩机制），同样还有一些用于维护请求。 long 33554432   高
compression.type producer用于压缩数据的压缩类型。默认是无压缩。正确的选项值是none、gzip、snappy。压缩最好用于批量处理，批量处理消息越多，压缩性能越好 string none   高
retries 设置大于0的值将使客户端重新发送任何数据，一旦这些数据发送失败。注意，这些重试与客户端接收到发送错误时的重试没有什么不同。允许重试将潜在的改变数据的顺序，如果这两个消息记录都是发送到同一个partition，则第一个消息失败第二个发送成功，则第二条消息会比第一条消息出现要早。 int 0   高
batch.size producer将试图批处理消息记录，以减少请求次数。这将改善client与server之间的性能。这项配置控制默认的批量处理消息字节数。不会试图处理大于这个字节数的消息字节数。发送到brokers的请求将包含多个批量处理，其中会包含对每个partition的一个请求。较小的批量处理数值比较少用，并且可能降低吞吐量（0则会仅用批量处理）。较大的批量处理数值将会浪费更多内存空间，这样就需要分配特定批量处理数值的内存大小。 int 16384   高
client.id 当向server发出请求时，这个字符串会发送给server。目的是能够追踪请求源头，以此来允许ip/port许可列表之外的一些应用可以发送信息。这项应用可以设置任意字符串，因为没有任何功能性的目的，除了记录和跟踪 string “”   中
connections.max.idle.ms 关闭连接空闲时间 long 540000   中
linger.ms producer组将会汇总任何在请求与发送之间到达的消息记录一个单独批量的请求。通常来说，这只有在记录产生速度大于发送速度的时候才能发生。然而，在某些条件下，客户端将希望降低请求的数量，甚至降低到中等负载一下。这项设置将通过增加小的延迟来完成–即，不是立即发送一条记录，producer将会等待给定的延迟时间以允许其他消息记录发送，这些消息记录可以批量处理。这可以认为是TCP种Nagle的算法类似。这项设置设定了批量处理的更高的延迟边界：一旦我们获得某个partition的batch.size，他将会立即发送而不顾这项设置，然而如果我们获得消息字节数比这项设置要小的多，我们需要“linger”特定的时间以获取更多的消息。 这个设置默认为0，即没有延迟。设定linger.ms=5，例如，将会减少请求数目，但是同时会增加5ms的延迟。 long 0   中
max.block.ms 控制block的时长,当buffer空间不够或者metadata丢失时产生block long 60000   中
max.request.size 请求的最大字节数。这也是对最大记录尺寸的有效覆盖。注意：server具有自己对消息记录尺寸的覆盖，这些尺寸和这个设置不同。此项设置将会限制producer每次批量发送请求的数目，以防发出巨量的请求。 int 1048576   中
partitioner.class 分区类 实现Partitioner 的class class 
org.
apache.
kafka.
clients.
producer.
internals.
DefaultPartitioner		   中
receive.buffer.bytes socket的接收缓存空间大小,当阅读数据时使用 int 32768   中
request.timeout.ms 客户端将等待请求的响应的最大时间,如果在这个时间内没有收到响应，客户端将重发请求;超过重试次数将抛异常 int 3000   中
send.buffer.bytes 发送数据时的缓存空间大小 int 131072   中
timeout.ms 此配置选项控制server等待来自followers的确认的最大时间。如果确认的请求数目在此时间内没有实现，则会返回一个错误。这个超时限制是以server端度量的，没有包含请求的网络延迟 int 30000   中
max.in.flight.requests.per.connection kafka可以在一个connection中发送多个请求，叫作一个flight,这样可以减少开销，但是如果产生错误，可能会造成数据的发送顺序改变,默认是5 (修改） int 5   低
metadata.fetch.timeout.ms 是指我们所获取的一些元素据的第一个时间数据。元素据包含：topic，host，partitions。此项配置是指当等待元素据fetch成功完成所需要的时间，否则会跑出异常给客户端 long 60000   低
metadata.max.age.ms 以微秒为单位的时间，是在我们强制更新metadata的时间间隔。即使我们没有看到任何partition leadership改变。 long 300000   低
metric.reporters 类的列表，用于衡量指标。实现MetricReporter接口，将允许增加一些类，这些类在新的衡量指标产生时就会改变。JmxReporter总会包含用于注册JMX统计 list none   低
metrics.num.samples 用于维护metrics的样本数 int 2   低
metrics.sample.window.ms metrics系统维护可配置的样本数量，在一个可修正的window size。这项配置配置了窗口大小，例如。我们可能在30s的期间维护两个样本。当一个窗口推出后，我们会擦除并重写最老的窗口 long 30000   低
reconnect.backoff.ms 连接失败时，当我们重新连接时的等待时间。这避免了客户端反复重连 long 10   低
retry.backoff.ms 在试图重试失败的produce请求之前的等待时间。避免陷入发送-失败的死循环中 long 100   低



Consumer配置

Property Default Description
group.id   用来唯一标识consumer进程所在组的字符串，如果设置同样的group id，表示这些processes都是属于同一个consumer group
zookeeper.connect   指定zookeeper的连接的字符串，格式是hostname：port，此处host和port都是zookeeper server的host和port，为避免某个zookeeper 机器宕机之后失联，你可以指定多个hostname：port，使用逗号作为分隔：hostname1：port1，hostname2：port2，hostname3：port3可以在zookeeper连接字符串中加入zookeeper的chroot路径，此路径用于存放他自己的数据，方式：hostname1：port1，hostname2：port2，hostname3：port3/chroot/path
consumer.id null 不需要设置，一般自动产生
socket.timeout.ms 30*1000 网络请求的超时限制。真实的超时限制是 max.fetch.wait+socket.timeout.ms
socket.receive.buffer.bytes 64*1024 socket用于接收网络请求的缓存大小
fetch.message.max.bytes 1024*1024 每次fetch请求中，针对每次fetch消息的最大字节数。这些字节将会督导用于每个partition的内存中，因此，此设置将会控制consumer所使用的memory大小。这个fetch请求尺寸必须至少和server允许的最大消息尺寸相等，否则，producer可能发送的消息尺寸大于consumer所能消耗的尺寸。
num.consumer.fetchers 1 用于fetch数据的fetcher线程数
auto.commit.enable true 如果为真，consumer所fetch的消息的offset将会自动的同步到zookeeper。这项提交的offset将在进程挂掉时，由新的consumer使用
auto.commit.interval.ms 60*1000 consumer向zookeeper提交offset的频率
queued.max.message.chunks 2 用于缓存消息的最大数目，以供consumption。每个chunk必须和fetch.message.max.bytes相同
rebalance.max.retries 4 当新的consumer加入到consumer group时，consumers集合试图重新平衡分配到每个consumer的partitions数目。如果consumers集合改变了，当分配正在执行时，这个重新平衡会失败并重入
fetch.min.bytes 1 每次fetch请求时，server应该返回的最小字节数。如果没有足够的数据返回，请求会等待，直到足够的数据才会返回。
fetch.wait.max.ms 100 如果没有足够的数据能够满足fetch.min.bytes，则此项配置是指在应答fetch请求之前，server会阻塞的最大时间。
rebalance.backoff.ms 2000 在重试reblance之前backoff时间
refresh.leader.backoff.ms 200 在试图确定某个partition的leader是否失去他的leader地位之前，需要等待的backoff时间
auto.offset.reset largest zookeeper中没有初始化的offset时，如果offset是以下值的回应：smallest：自动复位offset为smallest的offsetlargest：自动复位offset为largest的offsetanything else：向consumer抛出异常
consumer.timeout.ms -1 如果没有消息可用，即使等待特定的时间之后也没有，则抛出超时异常
exclude.internal.topics true 是否将内部topics的消息暴露给consumer
paritition.assignment.strategy range 选择向consumer 流分配partitions的策略，可选值：range，roundrobin
client.id group id value 是用户特定的字符串，用来在每次请求中帮助跟踪调用。它应该可以逻辑上确认产生这个请求的应用
zookeeper.session.timeout.ms 6000 zookeeper 会话的超时限制。如果consumer在这段时间内没有向zookeeper发送心跳信息，则它会被认为挂掉了，并且reblance将会产生
zookeeper.connection.timeout.ms 6000 客户端在建立通zookeeper连接中的最大等待时间
zookeeper.sync.time.ms 2000 ZK follower可以落后ZK leader的最大时间
offsets.storage zookeeper 用于存放offsets的地点： zookeeper或者kafka
offset.channel.backoff.ms 1000 重新连接offsets channel或者是重试失败的offset的fetch/commit请求的backoff时间
offsets.channel.socket.timeout.ms 10000 当读取offset的fetch/commit请求回应的socket 超时限制。此超时限制是被consumerMetadata请求用来请求offset管理
offsets.commit.max.retries 5 重试offset commit的次数。这个重试只应用于offset commits在shut-down之间。
dual.commit.enabled true 如果使用“kafka”作为offsets.storage，你可以二次提交offset到zookeeper(还有一次是提交到kafka）。在zookeeper-based的offset storage到kafka-based的offset storage迁移时，这是必须的。对任意给定的consumer group来说，比较安全的建议是当完成迁移之后就关闭这个选项
partition.assignment.strategy range 在“range”和“roundrobin”策略之间选择一种作为分配partitions给consumer 数据流的策略； 循环的partition分配器分配所有可用的partitions以及所有可用consumer 线程。它会将partition循环的分配到consumer线程上。如果所有consumer实例的订阅都是确定的，则partitions的划分是确定的分布。循环分配策略只有在以下条件满足时才可以：（1）每个topic在每个consumer实力上都有同样数量的数据流。（2）订阅的topic的集合对于consumer group中每个consumer实例来说都是确定的。
备注：从 0.9.0.0版本后，kafkat添加了新的消费者API及对应的consumer配置，有一些较大的变化，这就要求使用对应的kafka-clients(客户端)版本。






来自：csdn
作者;louisliao_1981

kafka0.9.0及0.10.0配置属性


联系
我们