Kafka常用配置说明

你的名字 2022-03-28 09:48 586阅读 0赞

**一.Broker配置**  
在conf/server.properties中进行设置  
[1.broker.id][](default 0)  
每个broker都需要有一个标识符，[建议host1.xx.com][host1.xx.com],[host2.xx.com][]  
2.listeners(default localhost:9092)  
监听端口，默认9092，如果使用1024以下端口，需要使用root权限启动Kafka。  
3.zookeeper.connect(default localhost:2181)  
用于保存broker元数据的Zookeeper地址，格式为hostname:port/path,其中path作为Kafka集群的chroot环境，默认为根目录  
4.log.dirs(default /tmp/kafka-logs)  
Kafka把所有消息都保存在磁盘上，用逗号分隔的本地文件系统路径，如果指定了多个路径，那么broker会根据"最少使用"原则，把同一个分区的日志片段保存在同一个路径下，且broker会往拥有最少数目分区的路径新增分区，而不是往拥有最小磁盘空间的路径新增分区。  
5.num.recovery.threads.per.data.dir(default 1)  
每个消息路径恢复的线程数，一般只有在3重情况下触发恢复:  
(1)服务器正常启动，用于打开每个分区的日志片段  
(2)服务器崩溃后重启，用于检查和截断每个分区的日志片段  
(3)服务器正常关闭，用于关闭日志片段  
6.auto.create.topics.enable(default true)  
是否支持自动创建主题,Kafka会在以下三种情况创建主题:  
(1)生产者开始往主题写入消息时  
(2)消费者开始从主题读取消息时  
(3)当任意一个客户端向主题发送元数据请求时。  
**二.Topic配置**  
在conf/server.properties中进行设置  
1.num.partitions(default 1)  
指定了新创建的主题将包含多少个分区，可以增加主题的分区个数，但不能减少分区的个数，如需创建分区个数少于num.partitions，那么就需要手动创建该主题。  
2.log.retention.hours,log.retention.minutes,[log.retention.ms][](default log.retention.hours=168=1周)  
Kafka根据这个设定的时间来决定数据可以保留多久  
3.log.retention.bytes(default 1G)  
通过保留的消息的字节数来判断消息是否过期,作用在每一个分区上。  
4.log.segment.bytes(default 1G)  
每个分区有段segment组成，限制segment的大小  
[5.segment.ms][](default 1周)  
指定segment过期时间  
6.message.max.bytes(default 1Mb)  
指定限制单个消息的大小

**生产者配置**  
1.bootstrap.servers  
节点服务器  
2.key.serializer  
键序列化  
3.value.serializer  
值序列化  
4.acks(default 1)  
指定有多少个节点收到信息就认为写入成功，ack=0表示不会等到节点服务器返回信息；ack=1表示首领节点收到信息即可；ack=all表示所有节点都接收到信息才确认成功  
5.buffer.memory(default 33554432=32Mb)  
设置生产者缓冲区大小，生产者用它缓存发送到服务器的消息。如果应用程序发送消息的速度大于发送到服务器的速度，那么会导致生产者空间不足。  
6.compression.type（default none）  
默认情况下消息不会被压缩，指定消息被发送给broker之前使用哪一种压缩算法进行压缩。可以设置为snappy,gzip,lz4。  
snappy:算法由google发明，占用较少的CPU，能提供较好的性能和相当可观的压缩比。  
gzip:一般会占用较多的CPU，但会提供更高的压缩比  
7.retries（default 2147483647）  
发生错误，生产者可以重发消息的次数，达到这个次数就放弃并返回错误  
8.batch.size(default 16384=16kb)  
当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里，该参数指定了一个批次可以使用的内存大小(注意:消费者并不会等到该内存满了才会发送)  
[9.linger.ms][](default 0)  
指定生产者发送批次之前等待更多消息加入批次的时间  
[10.client.id][]  
可以是任意字符串,broker会用它来识别消息的来源  
11.max.in.flight.requests.per.connection(default 5)  
指定生产者在收到服务器响应之前可以发送多少个消息，它的值越高就会占用更高的内存，不过也可以增加吞吐量。设置为1时可以保证按照发送的顺序写入服务器，即使发生了重试发送。  
[12.request.timeout.ms][](default 30000)  
生产者在发送数据时等待服务器返回响应的时间  
[13.max.block.ms][](default 60000)  
指定在调用send(),partitionsFor()方法获取元数据时生产者阻塞的时间，当生产者的缓冲区已满或者没有可用的元数据时，都会阻塞  
14.max.request.size(1048576=1Mb)  
用于控制生产者请求的大小。  
15.receive.buffer.bytes(32768=32kb)和send.buffer.bytes(131072=128kb)  
分别指定TCP socket接收和发送数据包的缓冲区大小，如果被设置为-1，则使用操作系统默认的大小。如果生产者或消费者与broker处于不同的数据中心，那么以适当增大这些值，因为跨数据中心的网络一般都有比较高的延迟和比较低的宽带。

**消费者配置**  
1.bootstrap.servers  
节点服务器  
2.key.serializer  
键序列化  
3.value.serializer  
值序列化  
[4.group.id][]  
消费者组ID  
5.fetch.min.bytes(default 1)  
指定消费者从服务器获取记录的最小字节数。broker在收到消费者数据请求时，如果可用数据量小于fetch.min.bytes,那么它会等到足够的可用数据才会返回给消费者。  
[6.fetch.max.wait.ms][](default 500)  
和fetch.min.bytes配合一起，如果等到fetch.max.wait.ms时间，而数据量没有达到fetch.min.bytes，那么也会将数据返回给消费者  
7.max.partition.fetch.bytes(default 1048576=1Mb)  
指定服务器从每个分区返回给消费者最大字节数，如果有一个主题20个分区，5个消费者，那么每个消费者需要4\*max.parttion.fetch.bytes的内存来接收记录  
[8.session.timeout.ms][](default 10000)  
指定消费者在被认为死亡之前可以与服务器断开连接的时间。[一般配合heartbeat.interval.ms][heartbeat.interval.ms],其指定了poll()方法发送心跳的频率，  
一般为session.timeout.ms时间的三分之一。  
9.auto.offset.reset(default latest)  
该属性指定消费者在读取一个没有偏移量的分区或者偏移量无效的情况下(因消费者长时间失效，包含偏移量的记录已经过时并被删除)该如何处理。  
latest表示在偏移量无效的情况下，消费者将从最新的记录开始读取数据  
earlest表示从起始位置读取分区记录  
10.enable.auto.commit(default true)  
指定消费者是否自动提交偏移量  
11.partition.assignment.strategy(default Range)  
分区分配给消费者的策略，有两种，Range,RoundRobin  
[12.client.id][]  
可以是任意字符串,broker会用它来识别消息的来源  
13.max.poll.records(default 500)  
单次调用poll方法能够返回的记录数量  
14.receive.buffer.bytes(65536=64kb)和send.buffer.bytes(131072=128kb)  
分别指定TCP socket接收和发送数据包的缓冲区大小，如果被设置为-1，则使用操作系统默认的大小。如果生产者或消费者与broker处于不同的数据中心，那么以适当增大这些值，因为跨数据中心的网络一般都有比较高的延迟和比较低的宽带。

[1.broker.id]: http://1.broker.id
[host1.xx.com]: http://xn--host1-gq2j979w.xx.com
[host2.xx.com]: http://host2.xx.com
[log.retention.ms]: http://log.retention.ms
[5.segment.ms]: http://5.segment.ms
[9.linger.ms]: http://9.linger.ms
[10.client.id]: http://10.client.id
[12.request.timeout.ms]: http://12.request.timeout.ms
[13.max.block.ms]: http://13.max.block.ms
[4.group.id]: http://4.group.id
[6.fetch.max.wait.ms]: http://6.fetch.max.wait.ms
[8.session.timeout.ms]: http://8.session.timeout.ms
[heartbeat.interval.ms]: http://xn--heartbeat-u75ny44c2t4ktdtb.interval.ms
[12.client.id]: http://12.client.id