文档中心 消息队列 CKafka 扩展阅读 CKafka 常用参数配置指南

CKafka 常用参数配置指南

最近更新时间:2019-04-15 15:39:29

Broker 配置参数说明

当前 CKafka broker 端的一些配置如下,供参考:

# 消息体的最大大小,单位是字节
message.max.bytes=1000012

# 是否允许自动创建 topic, 默认是 false,当前可以通过控制台或云 API 创建。 
auto.create.topics.enable=false

# 是否允许调用接口删除 Topic
delete.topic.enable=true

# Broker 允许的最大请求大小为 16MB 
socket.request.max.bytes=16777216

# 每个 IP 与 Broker 最多建立 5000 个连接
max.connections.per.ip=5000

# offset 保留时间,默认为 7 天
offsets.retention.minutes=10080

# 没有 ACL 设置时,允许任何人访问
allow.everyone.if.no.acl.found=true

# 日志分片大小为 1G
log.segment.bytes=1073741824

# 日志滚动检查间隔 5 分钟,当设置保留时间小于 5 分钟时,也可能需要等待 5 分钟才会清空日志。
log.retention.check.interval.ms=300000

注意:其他未列出的 Broker 配置参考 开源 Kafka 默认配置

Topic 配置参数说明

1. 选取合适的分区数量

从生产者的角度来看,向不同的 partition 写入是完全并行的;从消费者的角度来看,并发数完全取决于 partition 的数量(如果 consumer 数量大于 partition 数量,则必有 consumer 闲置)。因此选取合适的分区数量对于发挥 CKafka 实例的性能十分重要。
partition 的数量需要根据生产和消费的吞吐来判断。理想情况下,可以通过如下公式来判断分区的数目:

Num = max( T/PT , T/CT ) = T / min( PT , CT )

其中,Num 代表 partition 数量,T 代表目标吞吐量,PT 代表生产者写入单个 partition 的最大吞吐,CT 代表消费者从单个 partition 消费的最大吞吐。则 partition 数量应该等于 T/PT 和 T/CT 中较大的那一个。

在实际情况中,生产者写入但 partition 的最大吞吐 PT 的影响因素和批处理的规模、压缩算法、确认机制、副本数等有关。消费者从单个 partition 消费的最大吞吐 CT 的影响因素和业务逻辑有关,需要在不同场景下实测得出。

通常建议 partition 的数量一定要大于等于消费者的数量来实现最大并发。 如果消费者数量是 5,则 partition 的数目也应该是 >=5 的。同时,过多的分区会导致生产吞吐的降低和选举耗时的增加,因此也不建议过多分区。提供如下信息供参考:

  • 单个 partition 是可以实现消息的顺序写入的。
  • 单个 partition 只能被同消费者组的单个消费者进程消费。
  • 单个消费者进程可同时消费多个 partition,即 partition 限制了消费端的并发能力。
  • partition 越多则失败后 leader 选举的耗时越长。
  • offset 的粒度最细是在 partition 级别的,partition 越多,查询 offset 就越耗时。
  • partition 的数量是可以动态增加的,只能增加不能减少。但增加会出现消息 rebalance 的情况。

2. 选取合适的副本

目前为了保证可用性副本数必须大于等于 2,如果需要保障高可靠建议 3 副本。

注意:副本数会影响生产/消费流量,如 3 副本则实际流量= 生产流量*3

3. 日志保留时间

Topic 的 log.retention.ms 配置通过控制台实例的保留时间统一设置。

4. 其他 Topic 级别配置说明

# Topic 级别最大消息大小
max.message.bytes=1000012

# 0.10.2 版本消息格式为 V1 格式
message.format.version=0.10.2-IV0

# 不在 ISR 中的 replica 允许选择为 Leader,可用性高于可靠性,存在数据丢失风险。
unclean.leader.election.enable=true

# ISR 提交生产者请求的最小副本数。如果同步状态的副本数小于该值,服务器将不再接受。request.required.acks为-1或all的写入请求。
min.insync.replicas=1

生产者配置指南

生产端常用参数配置如下,建议客户根据实际业务场景调整配置:

# 生产者会尝试将业务发送到相同的 Partition 的消息合包发送到 Broker,batch.size 设置合包的大小上限。默认为 16K。batch.size 设太小会导致吞吐下降,设太大会导致内存使用过多。
batch.size=16384

# Kafka producer 的 ack 有 3 种机制,分别说明如下:
# -1 或 all:Broker 在 leader 收到数据并同步给所有 ISR 中的 follower 后,才应答给 Producer 继续发送下一条(批)消息。 这种配置提供了最高的数据可靠性,只要有一个已同步的副本存活就不会有消息丢失。注意:这种配置不能确保所有的副本读写入该数据才返回,可以配合 Topic 级别参数 min.insync.replicas 使用。
# 0:生产者不等待来自 broker 同步完成的确认,继续发送下一条(批)消息。这种配置生产性能最高,但数据可靠性最低(当服务器故障时可能会有数据丢失,如果 leader 已死但是 producer 不知情,则 broker 收不到消息)
# 1:生产者在 leader 已成功收到的数据并得到确认后再发送下一条(批)消息。这种配置是在生产吞吐和数据可靠性之间的权衡(如果leader已死但是尚未复制,则消息可能丢失)

# 用户不显示配置时,默认值为 1。用户根据自己的业务情况进行设置。
acks=1

# 控制生产请求在 Broker 等待副本同步满足 acks 设置的条件所等待的最大时间。
timeout.ms=30000

# 配置生产者用来缓存消息等待发送到 Broker 的内存。用户要根据生产者所在进程的内存总大小调节。
buffer.memory=33554432

# 当生产消息的速度比 Sender 线程发送到 Broker 速度快,导致 buffer.memory 配置的内存用完时会阻塞生产者 send 操作,该参数设置最大的阻塞时间。
max.block.ms=60000

# 客户端在每个连接上最多可发送的最大的未确认请求数,该参数大于 1 且 retries 大于 0 时可能导致数据乱序。
max.in.flight.requests.per.connection=5

# 设置消息延迟发送的时间,这样可以等待更多的消息组成 batch 发送。默认为 0 表示立即发送。当待发送的消息达到 batch.size 设置的大小时,不管是否达到 linger.ms 设置的时间,请求也会立即发送。
linger.ms=0

# 生产者能够发送的请求包大小上限,默认为 1M。在修改该值时注意不能超过 Broker 配置的包大小上限 16M。
max.request.size=1048576

# 压缩格式配置,目前 0.9(包含)以下版本不允许使用压缩,0.10(包含)以上不允许使用 GZip 压缩。
compression.type=[none, snappy, lz4]

# 客户端发送给 Broker 的请求的超时时间,不能小于 Broker 配置的 replica.lag.time.max.ms,目前该值为 10000ms。
request.timeout.ms=30000

# 请求发生错误时重试次数,当 retries 大于 0 且 max.in.flight.requests.per.connection 大于 1 时可能会导致乱序。建议客户将该值改成大于 0,即重复发送失败的消息。
retries=0

# 发送请求失败时到下一次重试请求之间的时间。
retry.backoff.ms=100

消费者配置指南

生产端常用参数配置如下,建议客户根据实际业务场景调整配置:

# 是否在消费消息后将 offset 同步到 Broker,当 Consumer 失败后就能从 Broker 获取最新的 offset。
auto.commit.enable=true

# 当 auto.commit.enable=true 时,自动提交 Offset 的时间间隔,建议设置至少 1000。
auto.commit.interval.ms=5000

# 当 Broker 端没有 offset(如第一次消费或 offset 超过 7 天过期)时如何初始化 offset,当收到 OFFSET_OUT_OF_RANGE 错误时,如何重置 Offset。
# earliest:表示自动重置到 partition 的最小 offset。
# latest:默认为 latest,表示自动重置到 partition 的最大 offset。
# none:不自动进行 offset 重置,抛出 OffsetOutOfRangeException 异常。
auto.offset.reset=latest

# 标识消费者所属的消费分组
group.id=""

# 使用 Kafka 消费分组机制时,消费者超时时间。当 Broker 在该时间内没有收到消费者的心跳时,认为该消费者故障失败,Broker 发起重新 Rebalance 过程。目前该值的配置必须在 Broker 配置。group.min.session.timeout.ms=6000和group.max.session.timeout.ms=300000 之间
session.timeout.ms=10000

# 使用 Kafka 消费分组机制时,消费者发送心跳的间隔。这个值必须小于 session.timeout.ms,一般小于它的三分之一
heartbeat.interval.ms=3000

# 使用 Kafka 消费分组机制时,再次调用 poll 允许的最大间隔。如果在该时间内没有再次调用 poll,则认为该消费者已经失败,Broker 会重新发起 Rebalance 把分配给它的 partition 分配给其他消费者
max.poll.interval.ms=300000

# Fecth 请求最少返回的数据大小。默认设置为 1B,表示请求能够尽快返回。增大该值会增加吞吐,同时也会增加延迟
fetch.min.bytes=1

# Fetch 请求最多返回的数据大小,默认设置为 50MB
fetch.max.bytes=52428800

# Fetch 请求等待时间
fetch.max.wait.ms=500

# Fetch 请求每个 partition 返回的最大数据大小,默认为 10MB
max.partition.fetch.bytes=1048576

# 在一次 poll 调用中返回的记录数
max.poll.records=500

# 客户端请求超时时间,如果超过这个时间没有收到应答,则请求超时失败
request.timeout.ms=305000