文章目录
- Leader Partition负载均衡
- 消费者初始化流程
- 消费者再平衡
- 生产者和消费者如何提高吞吐量
- 如何发送大消息
Leader Partition负载均衡
参数名称 | 描述 |
---|---|
auto.leader.rebalance.enable | 默认是true。自动LeaderPartition平衡。生产环境中,leader重选举的代价比较大,可能会带来性能影响,建议设置为false关闭 |
leader.imbalance.per.broker.percentage | 默认是10%。每个broker允许的不平衡的leader的比率。如果每个broker超过了这个值,控制器会触发leader的平衡。 |
leader.imbalance.check.interval.seconds | 默认值300秒。检查leader负载是否平衡的间隔时间。 |
leader 不平衡的比例是如何计算呢?
当leader发生过宕机后,kafka投票重新选举新的leader,这种情况下就会导致leader和第一个副本节点不一致
,则不平衡数+1。根据公式,平衡率2/4= 50%,大于默认的平衡率10%,会自动触发自动平衡。
本人试了一下,即时在server.properties中配置了auto.leader.rebalance.enable=true,集群重启依然会导致leader不平衡问题
消费者初始化流程
参数 | 描述 |
---|---|
bootstrap.servers | 向Kafka集群建立初始连接用到的host/port列表。 |
key.deserializer value.deserializer | 指定接收消息的key和value的反序列化类型。一定要写全类名。 |
group.id | 标记消费者所属的消费者组。 |
enable.auto.commit | 默认值为true ,消费者会自动周期性地向服务器提交偏移量。 |
auto.commit.interval.ms | 如果设置了enable.auto.commit的值为true,则该值定义了消费者偏移量向Kafka提交的频率,默认5s 。 |
auto.offset.reset | 当Kafka中没有初始偏移量或当前偏移量在服务器中不存在(如,数据被删除了),该如何处理?earliest:自动重置偏移量到最早的偏移量。latest:默认,自动重置偏移量为最新的偏移量 。none:如果消费组原来的(previous)偏移量不存在,则向消费者抛异常。anything:向消费者抛异常。 |
offsets.topic.num.partitions | __consumer_offsets的分区数,默认是50个分区。不建议修改 。 |
heartbeat.interval.ms | Kafka消费者和coordinator之间的心跳时间,默认3s 。该条目的值必须小于session.timeout.ms,也不应该高于session.timeout.ms的1/3。不建议修改 。 |
session.timeout.ms | Kafka消费者和coordinator之间连接超时时间,默认45s。超过该值,该消费者被移除,消费者组执行再平衡 |
max.poll.interval.ms | 消费者处理消息的最大时长,默认是5分钟。超过该值,该消费者被移除,消费者组执行再平衡。 |
fetch.min.bytes | 默认1个字节 。消费者获取服务器端一批消息最小的字节数。 |
fetch.max.wait.ms | 默认500ms 。如果没有从服务器端获取到一批数据的最小字节数。该时间到,仍然会返回数据。 |
fetch.max.bytes | 默认Default:52428800(50 m )。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值(50m)仍然可以拉取回来这批数据,因此,这不是一个绝对最大值。一批次的大小受message.max.bytes(brokerconfig)or max.message.bytes(topic config)影响 |
max.poll.records | 一次poll拉取数据返回消息的最大条数,默认是500条。 |
消费者再平衡
参数 | 说明 |
---|---|
heartbeat.interval.ms | Kafka消费者和coordinator之间的心跳时间,默认3s。该条目的值必须小于session.timeout.ms,也不应该高于session.timeout.ms的1/3。 |
session.timeout.ms | Kafka消费者和coordinator之间连接超时时间,默认45s。超过该值,该消费者被移除,消费者组执行再平衡。 |
max.poll.interval.ms | 消费者处理消息的最大时长,默认是5分钟。超过该值,该消费者被移除,消费者组执行再平衡。 |
partition.assignment.strategy | 消费者分区分配策略,默认策略是Range +CooperativeSticky 。Kafka可以同时使用多个分区分配策略。可以选择的策略包括:Range、RoundRobin、Sticky、 |
CooperativeSticky |
生产者和消费者如何提高吞吐量
1)提升生产吞吐量
(1)buffer.memory:发送消息的缓冲区大小,默认值是32m,可以增加到64m。
(2)batch.size:默认是16k。如果batch设置太小,会导致频繁网络请求,吞吐量下降;如果batch太大,会导致一条消息需要等待很久才能被发送出去,增加网络延时。
(3)linger.ms,这个值默认是0,意思就是消息必须立即被发送。一般设置一个5-100毫秒。如果linger.ms设置的太小,会导致频繁网络请求,吞吐量下降;如果linger.ms太长,会导致一条消息需要等待很久才能被发送出去,增加网络延时。
(4)compression.type:默认是none,不压缩,但是也可以使用lz4压缩,效率还是不错的,压缩之后可以减小数据量,提升吞吐量,但是会加大producer端的CPU开销。
2)增加分区3)消费者提高吞吐量
(1)调整fetch.max.bytes大小,默认是50m。
(2)调整max.poll.records大小,默认是500条。
4)增加下游消费者处理能力
如何发送大消息
有些场景需要kafka发送很大的消息,比如消息中含有base64格式的图片或短视频信息等,这就需要发送大消息了
参数 | 描述 | 备注 |
---|---|---|
message.max.bytes | 默认1m ,broker端接收每个批次消息最大值。 | broker配置server.properties,例如该设置为20000000(20M) |
replica.fetch.max.bytes | 默认1m ,副本同步数据,每个批次消息最大值。 | broker配置server.properties,例如设置为21000000(21M),这是单个消息的最大值broker可复制的消息的最大字节数,比message.max.bytes大,否则broker会接收此消息,但无法将此消息复制出去,从而造成数据丢失,但是不能太大!!!不然复制的时候容易内存溢出 |
max.request.size | 默认1m ,生产者发往broker每个请求消息最大值。针对topic级别设置消息体的大小。例如设置为20000000(20M), 这是生产者能请求的最大消息,大于或等于message.max.bytes | 生产者配置 |
fetch.max.bytes | 默认:52428800(50m )。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值(50m)仍然可以拉取回来这批数据,因此,这不是一个绝对最大值。一批次的大小受message.max.bytes(broker config)or max.message.bytes(topic config)影响。 例如设置为 21000000(21M), 这是消费者能读取的最大消息,大于或等于message.max.bytes | 消费者配置,consumer.properties中添加 |
yml或者propertise中配置max.request.size,但实测不生效,需要java中写config,具体的配置可在https://kafka.apache.org/documentation/#producerconfigs中查找,如下进行java的kafka配置:
@Bean
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.10.100.100:9092");
props.put(ProducerConfig.RETRIES_CONFIG, 3);
//每次批量发送消息的数量(吞吐量和延时性能),默认16k,batch 越小,producer的吞吐量越低,
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); //默认32M
props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 50000000);
//# 指定消息key和消息体的编解码方式
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}