Kafka 分区分配及再均衡战略深度剖析与斲丧者事件和数据积蓄的简单先容

Kafka：分布式消息体系的核心原理与安装摆设-CSDN博客
自界说 Kafka 脚本 kf-use.sh 的剖析与功能与应用示例-CSDN博客
Kafka 生产者全面剖析：从根本原理到高级实践-CSDN博客
Kafka 生产者优化与数据处理处罚履历-CSDN博客
Kafka 工作流程剖析：从 Broker 工作原理、节点的服役、退役、副本的天生到数据存储与读写优化-CSDN博客
Kafka 斲丧者全面剖析：原理、斲丧者 API 与Offset 位移-CSDN博客
Kafka 分区分配及再均衡战略深度剖析与斲丧者事件和数据积蓄的简单先容-CSDN博客
Kafka 数据倾斜：缘故原由、影响与办理方案-CSDN博客
Kafka 核心要点剖析_kafka mirrok-CSDN博客
Kafka 核心题目深度剖析：全面明白分布式消息队列的关键要点_kafka队列日志-CSDN博客

目次
一、分区分配战略根本
二、Range 分区分配战略
（一）原理
（二）案例
（三）Range 分区分配再均衡案例
三、RoundRobin 分区分配战略
（一）原理
（二）案例
（三）RoundRobin 分区分配再均衡案例
四、Sticky 分区分配战略
（一）原理
（二）案例
（三）Sticky 分区分配再均衡案例
五、CooperativeSticky 分区分配战略
六、斲丧者事件
七、数据积蓄（斲丧者怎样进步吞吐量）
八、总结

---

         在 Kafka 的斲丧使命处理处罚中，分区的分配以及再均衡是至关告急的环节。公道的分区分配战略可以或许确保斲丧者高效地处理处罚消息，而明白再均衡机制则有助于应对斲丧者组在运行过程中的动态厘革。本文将深入探究 Kafka 中差别的分区分配战略，包罗 Range、RoundRobin、Sticky 和 CooperativeSticky，以及它们在各种场景下的再均衡体现，并团结现实案例举行具体分析，并对斲丧者事件和数据积蓄举行简单先容。

一、分区分配战略根本

         在一个 Kafka 斲丧者组中，包罗多个斲丧者，而一个主题则由多个分区构成。关键题目在于确定哪个斲丧者来斲丧哪个分区的数据。Kafka 提供了四种主流的分区分配战略，而且可以通过设置参数 partition.assignment.strategy 来修改分区的分配战略，默认战略是 Range + CooperativeSticky。同时，另有一些相干的告急参数：

参数名称	形貌
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

二、Range 分区分配战略


（一）原理


         Range 分区分配战略是基于主题的分区数目和斲丧者数目举行分配。它会按照序次将连续的分区分配给每个斲丧者，尽大概均匀地分配分区，但大概会导致差别斲丧者分配到的分区数目不划一。

（二）案例


起首，将主题 first 修改为 7 个分区：

1. bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --alter --topic first --partitions 7

复制代码

注意，分区数可增长但不能淘汰，主题的副本数修改必要订定筹划实行，不能直接修改。

          由三个斲丧者 CustomConsumer、CustomConsumer1、CustomConsumer2 构成斲丧者组，组名都为 “test”，同时启动这 3 个斲丧者。

         启动 CustomProducer 生产者，发送 500 条消息，随机发送到差别的分区（修改发送次数为 500 次）。

1. package com.bigdata.kafka.producer;
3. import org.apache.kafka.clients.producer.*;
4. import org.apache.kafka.common.serialization.StringSerializer;
6. import java.util.Properties;
8. public class CustomProducerCallback {
9.     public static void main(String[] args) throws InterruptedException {
10.         // 1. 创建 kafka 生产者的配置对象
11.         Properties properties = new Properties();
12.         // 2. 给 kafka 配置对象添加配置信息
13.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
14.                 "192.168.235.128:9092");
15.         // key,value 序列化（必须）：key.serializer，value.serializer
16.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
17.                 StringSerializer.class.getName());
19.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
20.                 StringSerializer.class.getName());
21.         // 3. 创建 kafka 生产者对象
22.         KafkaProducer<String, String> kafkaProducer = new
23.                 KafkaProducer<String, String>(properties);
24.         // 4. 调用 send 方法,发送消息
25.         for (int i = 0; i < 500; i++) {
26.             // 添加回调
27.             kafkaProducer.send(new ProducerRecord<>("first",
28.               

复制代码