Kafka 0.8 配置参数解析

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
全局流量管理 GTM,标准版 1个月
云原生网关 MSE Higress,422元/月
简介:

Broker Configs

4个必填参数,

broker.id 
Each broker is uniquely identified by a non-negative integer id 
broker唯一标识,broker可以在不同的host或port,但必须保证id唯一

log.dirs (/tmp/kafka-logs) 
日志文件存放的目录 
可以用逗号隔开多个目录,当创建partitions时,会自动挑一个已创建partition最少的目录创建 
因为Kafka必须充分利用磁盘资源,所以要让partititons均匀分布在所有disks上,至少每个disk创建一个目录

port (6667) 
broker server所在端口

zookeeper.connect 
zk的地址,hostname1:port1,hostname2:port2

 

可选的重要参数,

advertised.host.name (null) 
设置暴露给其他Consumer,producer,broker的域名 
不设置会暴露默认域名,域名解析比较麻烦,这里可以设为ip

message.max.bytes (1000000) 
broker可以接收的message的最大size 
注意要和consumer的maximum fetch size相匹配

num.io.threads (8) 
处理I/O的线程数,即读写磁盘的线程数,所以至少等于磁盘数

queued.max.requests (500) 
可以queue的最大请求数,包括producer的写请求和consumer的读请求

log.segment.bytes (1024 * 1024 * 1024 
log.roll.hours 
上面2个配置用于设置,何时为partition产生新的segment文件 
为了防止partition文件过大,所以partition是由一组segment文件组成 
可以通过设置size或时间来决定何时roll

log.retention.{minutes,hours} (24 * 7) 
log.retention.bytes (-1) 
log.retention.check.interval.ms (5*60*1000, 5分钟) 
默认是7天数据 
当然你可以通过partition大小来设置删除threshold,默认是-1,即关掉的,注意这里设置的是partition大小,如果多个partition需要乘上partition数

log.cleanup.policy (delete or compact, default delete) 
log.cleaner.enable (false) 
log.cleaner.threads (1) 
当前retention支持两种类型,默认就是删除旧数据 
新加了compact模式,应用场景来源于Linkedin之前的Databus 
上面列出主要的相关配置,如果要使用compact 
log.cleanup.policy = compact 
log.cleaner.enable  = true 
http://kafka.apache.org/documentation.html#compaction 
这个场景,数据流中的key是有限重复的,只是在不断的更新value 
compact过程就是当触发retention条件时,删除相同key的旧值,只保留最新值

image 
局限, 
a. 无法指定compact的范围,除了last segment(正在被写),其他所有的segments都可能被compact 
b. 不支持compress topics

 

log.index.size.max.bytes (10*1024*1024) 
log.index.interval.bytes (4096) 
这个由于0.8版本,做出妥协,用逻辑offset来替换原先的物理offset 
那就需要在逻辑offset和物理地址之间创建index 
log.index.size.max.bytes,一个segment最大的index的大小,超出会创建新的segment,无论segment本身的配置 
log.index.interval.bytes,做index的间隔,也是系统为了找到一个物理offset需要做linear scan的最大值

 

log.flush.interval.messages (None) 
log.flush.interval.ms (None) 
log.flush.scheduler.interval.ms (3000),多久check 
为了效率,broker会缓存部分log,然后再flush到磁盘 
这里可以通过message数或时间来控制flush策略

 

replica相关的参数,

default.replication.factor (1), The default replication factor for automatically created topics 
默认的副本数

replica.lag.time.max.ms (10000), follower的最大lag时间,即如果leader在这个时间内都没有收到follower的fetch请求,就认为它dead 
If a follower hasn't sent any fetch requests for this window of time, the leader will remove the follower from ISR (in-sync replicas) and treat it as dead. 
replica.lag.max.messages (4000), 最大lag消息数,超过这个消息数,leader会认为该follower dead 
If a replica falls more than this many messages behind the leader, the leader will remove the follower from ISR and treat it as dead.

replica.socket.timeout.ms (30 * 1000), The socket timeout for network requests to the leader for replicating data. 
replica.socket.receive.buffer.bytes (64 * 1024), The socket receive buffer for network requests to the leader for replicating data.

replica.fetch.max.bytes (1024 * 1024), 一次fetch request最大可以fetch的数据size 
The number of byes of messages to attempt to fetch for each partition in the fetch requests the replicas send to the leader. 
replica.fetch.wait.max.ms (500), fetch request的等待超时 
The maximum amount of time to wait time for data to arrive on the leader in the fetch requests sent by the replicas to the leader. 
replica.fetch.min.bytes (1), Minimum bytes expected for each fetch response for the fetch requests from the replica to the leader. If not enough bytes, wait up to replica.fetch.wait.max.ms for this many bytes to arrive.

num.replica.fetchers (1), follower上用于同步leader数据的线程数 
Number of threads used to replicate messages from leaders. Increasing this value can increase the degree of I/O parallelism in the follower broker.

replica.high.watermark.checkpoint.interval.ms (5000), checkpoint HW的interval 
The frequency with which each replica saves its high watermark to disk to handle recovery.

 

failover相关的参数,

controlled.shutdown.enable (false) 
用于Graceful shutdown,当设为true时,在用stop命令关闭broker, 
会做下面两点优化, 
a. sync all its logs to disk,这样可以加速重启,避免checksum校验等步骤 
b. migrate任意partitions的leader replica到其他的brokers,这样主动的迁移,比发现异常后被动迁移更快 
注意,设为true,只有当broker上所有的replica,不是该partition唯一replica时才能stop成功 
因为如果直接关掉,会导致该partition不work

auto.leader.rebalance.enable (false) 
leader.imbalance.per.broker.percentage (10) 
leader.imbalance.check.interval.seconds (300)
 
broker crash后,partition的leader replica会被迁到其他的broker 
但是当broker恢复后,上面的replica会成为follower,但不会自动切成leader 
这样会导致不均衡,所以当设为true时,会自动做leader的rebalance 
默认超过10%,认为不均衡,300秒check一次

 

Topic-level configuration

其中一些参数是可以按topic设置的,并且可以修改删除

> bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic my-topic --partitions 1 
        --replication-factor 1 --config max.message.bytes=64000 --config flush.messages=1
> bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my-topic 
    --config max.message.bytes=128000
> bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my-topic 
    --deleteConfig max.message.bytes

上面分别是创建,修改和删除的命令 
可以按topic设置的参数包括,参考文档 
http://kafka.apache.org/documentation.html#topic-config

需要注意的是,这里的属性名和配置文件中的属性名是不一样的。。。

 

Producer Configs

参考,Kafka Producer接口

 

Consumer Configs

首先只有在用high level consumer时才需要管这个配置 
如果用low level,自己做代码里面写,这个配置不起作用的

2个必填参数,

group.id 
A string that uniquely identifies the group of consumer processes to which this consumer belongs.

zookeeper.connect 
zk地址,hostname1:port1,hostname2:port2,hostname3:port3/chroot/path 
可以看到zk地址其实是可以设置到,目录级别的(比如用于一个zk管理多个kafka集群),但是如果不用默认目录,需要自己创建和维护这个zk目录 
并且必须要和broker中的zk配置保持一致

 

可选配置,

fetch.message.max.bytes (1024 * 1024) 
The number of byes of messages to attempt to fetch for each topic-partition in each fetch request 
为了效率,consumer从kafka fetch数据,也是批量fetch的,虽然你处理的时候是onebyone逻辑,但后台是预读的 
这个值至少要等于broker里面设置的maximum message size,否则有可能连一条message都取不下来 
默认是1m,设太大会爆内存,而且会增大读重的可能性,因为当consumer发生变化时,会发生rebalance,这时被新分配到这个partition的consumer仍然会读到预读但没有commit的数据

auto.commit.enable (true), auto.commit.interval.ms (60 * 1000) 
默认是会自动commit offset到zk的,如果要自己commit,设为false 
默认除非自动commit的时间是1分钟 
用手工commit的简单的理由是,防止consumer crash丢失数据,可以在确认数据被正确处理后,再手工commit offset

queued.max.message.chunks (10) 
上面说的fetch.message.max.bytes是针对一个partition的,但是一个consumer是可以同时读多个partition,所以对每个partition都可以读一个这样的chunk 
所以这个配置是,同时可以读多少大小为fetch.message.max.bytes的chunk 
默认是10,个人认为至少要大于读的partition个数吧,但避免耗尽内存

fetch.wait.max.ms (100), fetch.min.bytes (1) 
当取不到fetch.message.max.bytes时,最大block时间,并且返回fetch.min.bytes大小数据

auto.offset.reset 
* smallest : automatically reset the offset to the smallest offset 
* largest : automatically reset the offset to the largest offset 
* anything else: throw exception to the consumer.

consumer.timeout.ms (-1) 
Throw a timeout exception to the consumer if no message is available for consumption after the specified interval 
默认是block的,当consumer取不到数据



本文章摘自博客园,原文发布日期:2014-09-04

目录
相关文章
|
15天前
|
Java
解析Java线程池:参数详解与执行流程
解析Java线程池:参数详解与执行流程
13 1
|
13天前
|
存储 缓存 运维
云计算中的服务器选型与配置:技术深度解析
【6月更文挑战第29天】云计算服务器选型与配置深度解析:关注业务需求、技术要求及成本效益。重点包括CPU、内存、存储和网络配置的优化,结合负载均衡、缓存、虚拟化和自动化运维策略,以提升性能和效率,确保云服务的稳定与高效。
|
13天前
|
消息中间件 Kafka 程序员
Kafka面试必备:深度解析Replica副本的作用与机制
**Kafka的Replica副本是保证数据可靠性的关键机制。每个Partition有Leader和Follower副本,Leader处理读写请求及管理同步,Follower被动同步并准备成为新Leader。从Kafka 2.4开始,Follower在完全同步时也可提供读服务,提升性能。数据一致性通过高水位机制和Leader Epoch机制保证,后者更精确地判断和恢复数据一致性,增强系统容错能力。**
22 1
|
15天前
|
消息中间件 监控 Kafka
深入解析:Kafka 为何不支持全面读写分离?
**Kafka 2.4 引入了有限的读写分离,允许Follower处理只读请求,以缓解Leader压力。但这不适用于所有场景,特别是实时数据流和日志分析,因高一致性需求及PULL同步方式导致的复制延迟,可能影响数据实时性和一致性。在设计系统时需考虑具体业务需求。**
14 1
|
15天前
|
域名解析 网络协议 Linux
Linux系统下DNS配置指南
Linux系统下DNS配置指南
27 1
|
8天前
|
传感器 编解码 API
【STM32开发入门】温湿度监测系统实战:SPI LCD显示、HAL库应用、GPIO配置、UART中断接收、ADC采集与串口通信全解析
SPI(Serial Peripheral Interface)是一种同步串行通信接口,常用于微控制器与外围设备间的数据传输。SPI LCD是指使用SPI接口与微控制器通信的液晶显示屏。这类LCD通常具有较少的引脚(通常4个:MISO、MOSI、SCK和SS),因此在引脚资源有限的系统中非常有用。通过SPI协议,微控制器可以向LCD发送命令和数据,控制显示内容和模式。
|
9天前
|
消息中间件 NoSQL Kafka
日志收集平台项目nginx、kafka、zookeeper、filebeat搭建的基本配置(2)
日志收集平台项目nginx、kafka、zookeeper、filebeat搭建的基本配置(2)
|
9天前
|
消息中间件 应用服务中间件 Kafka
日志收集平台项目nginx、kafka、zookeeper、filebeat搭建的基本配置(1)
日志收集平台项目nginx、kafka、zookeeper、filebeat搭建的基本配置(1)
|
9天前
|
消息中间件 Kafka
kafka配置中启动zookeeper时没有启动成功的解决办法
kafka配置中启动zookeeper时没有启动成功的解决办法
|
11天前
|
JavaScript
js 获取并解析 url 中参数的三种方法
js 获取并解析 url 中参数的三种方法
9 0

推荐镜像

更多