kafka rebalance 机制与Consumer多种消费模式案例应用实战-kafka 商业环境实战
本套系列博客从真实商业环境抽取案例进行总结和分享，并给出Spark商业应用实战指导，请持续关注本套博客。版权声明：本套Spark商业应用实战归做者（秦凯新）全部，禁止转载，欢迎学习。java
- kafka 商业环境实战-kafka生产环境规划
- kafka 商业环境实战-kafka生产者和消费者吞吐量测试
- kafka 商业环境实战-kafka生产者Producer参数设置及参数调优建议
- kafka 商业环境实战-kafka集群管理重要操做指令运维兵书
- kafka 商业环境实战-kafka集群Broker端参数设置及调优准则建议
- kafka 商业环境实战-kafka之Producer同步与异步消息发送及事务幂等性案例应用实战
- kafka 商业环境实战-kafka Poll轮询机制与消费者组的重平衡分区策略剖析
- kafka 商业环境实战-kafka rebalance 机制与Consumer多种消费模式案例应用实战
- [kafka 商业环境实战-kafka副本与ISR同步机制原理深刻剖析]
- [kafka 商业环境实战-kafka精确一次语义EOS的原理深刻剖析]
- [kafka 商业环境实战-kafka消息的幂等性与事务支持机制深刻剖析]
- [kafka 商业环境实战-kafka集群Controller竞选与责任设计思路架构详解]
- [kafka 商业环境实战-kafka集群消息格式之V1版本到V2版本的平滑过渡详解]
- [kafka 商业环境实战-kafka集群水印与leader epoch对数据一致性保障的深刻研究]
- [kafka 商业环境实战-kafka集群日志文件系统设计与留存机制及Compact深刻研究]
- [kafka 商业环境实战-kafka集群Consumer group状态机及Coordinaor管理机制深刻剖析]
- [kafka 商业环境实战-kafka调优过程在吞吐量，持久性，低延时，可用性等指标的折中选择研究]
1 rebalance 什么时候触发？到底干吗？流程如何？
1.1 reblance 什么时候触发
- 组订阅发生变动，好比基于正则表达式订阅，当匹配到新的topic建立时，组的订阅就会发生变动。
- 组的topic分区数发生变动，经过命令行脚本增长了订阅topic的分区数。
- 组成员发生变动：新加入组以及离开组。
1.2 reblance 到底干吗
一句话：多个Consumer订阅了一个Topic时，根据分区策略进行消费者订阅分区的重分配正则表达式
1.3 Coordinator 到底在那个Broker
找到Coordinator的算法 与 找到_consumer_offsets目标分区的算法是一致的。算法
- 第一步：肯定目标分区：Math.abs(groupId.hashCode)%50，假设是12。
- 第二步：找到_consumer_offsets分区为10的Leader副本所在的Broker，那么该broker即为Group Coordinator。
1.4 reblance 流程如何
reblance 流程流程总体以下图所示，值得强调的几点以下：apache
- Coordinator的角色由Broker端担任。
- Group Leader 的角色主要有Consumer担任。
- 加入组请求（JoinGroup）=>做用在于选择Group Leader。
- 同步组请求（SyncGroup）=>做用在于肯定分区分配方案给Coordinator，把方案响应给全部Consumer。
1.5 reblance 机制的好处
- 分区分配权利下放给客户端consumer，所以系统不用重启，既能够实现分区策略的变动。
- 用户能够自行实现机架感知分配方案。
1.6 reblance generation 过滤无用请求
- kafka引入 reblance generation ，就是为了防止Consumer group的无效Offset提交。若由于某些缘由，consumer延迟提交了Offset，而该consumer被踢出了消费组，那么该Consumer再次提交位移时，携带的就是旧的generation了。
2 reblance 监听器应用级别实战
-
reblance 监听器解决用户把位移提交到外部存储的状况，在监听器中实现位移保存和位移的重定向。bootstrap
-
onPartitionsRevoked : rebalance开启新一轮的重平衡前会调用，通常用于手动提交位移，及审计功能数组
-
onPartitionsAssigned ：rebalance在重平衡结束后会调用，通常用于消费逻辑处理session
Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test"); props.put("enable.auto.commit", "false"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); 统计rebalance总时长 final AtomicLong totalRebalanceTimeMs =new AtomicLong(0L) 统计rebalance开始时刻 final AtomicLong rebalanceStart =new AtomicLong(0L) 1 重平衡监听 consumer.subscribe(Arrays.asList("test-topic"), new ConsumerRebalanceListener(){ @Override public void onPartitionsRevoked(Collection<TopicPartition> partitions) { for(TopicPartition tp : partitions){ 1 保存到外部存储 saveToExternalStore(consumer.position(tp)) 2 手动提交位移 //consumer.commitSync(toCommit); } rebalanceStart.set(System.currentTimeMillis()) } } @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) { totalRebalanceTimeMs.addAndGet(System.currentTimeMillis()-rebalanceStart.get()) for (TopicPartition tp : partitions) { consumer.seek(tp，readFromExternalStore(tp)) } } }); 2 消息处理 while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) { buffer.add(record); } if (buffer.size() >= minBatchSize) { insertIntoDb(buffer); consumer.commitSync(); buffer.clear(); } } 复制代码
3 Consumer组内消息均衡实战
3.1 Consumer 单线程封装，实现多个消费者来消费（浪费资源）
实例主题：多线程
- ConsumerGroup 实现组封装
- ConsumerRunnable 每一个线程维护私有的KafkaConsumer实例
public class Main {
public static void main(String[] args) {
String brokerList = "localhost:9092";
String groupId = "testGroup1";
String topic = "test-topic";
int consumerNum = 3;
核心对外封装
ConsumerGroup consumerGroup = new ConsumerGroup(consumerNum, groupId, topic, brokerList);
consumerGroup.execute();
}
}
复制代码
import java.util.ArrayList;
import java.util.List;
public class ConsumerGroup {
private List<ConsumerRunnable> consumers;
public ConsumerGroup(int consumerNum, String groupId, String topic, String brokerList) {
consumers = new ArrayList<>(consumerNum);
for (int i = 0; i < consumerNum; ++i) {
ConsumerRunnable consumerThread = new ConsumerRunnable(brokerList, groupId, topic);
consumers.add(consumerThread);
}
}
public void execute() {
for (ConsumerRunnable task : consumers) {
new Thread(task).start();
}
}
}
复制代码
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
public class ConsumerRunnable implements Runnable {
private final KafkaConsumer<String, String> consumer;
public ConsumerRunnable(String brokerList, String groupId, String topic) {
Properties props = new Properties();
props.put("bootstrap.servers", brokerList);
props.put("group.id", groupId);
props.put("enable.auto.commit", "true"); //本例使用自动提交位移
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
this.consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList(topic)); // 本例使用分区副本自动分配策略
}
@Override
public void run() {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(200);
for (ConsumerRecord<String, String> record : records) {
System.out.println(Thread.currentThread().getName() + " consumed " + record.partition() +
"th message with offset: " + record.offset());
}
}
}
}
复制代码
3.2 一个Consumer，内部实现多线程消费（consumer压力过大）
实例主题：架构
- ConsumerHandler 单一的Consumer实例，poll后里面会跑一个线程池，执行多个Processor线程来处理
- Processor 业务逻辑处理方法
进一步优化建议；运维
- ConsumerHandler 设置手动提交位移，负责最终位移提交consumer.commitSync();。
- ConsumerHandler设置一个全局的Map<TopicPartion,OffsetAndMetadata> offsets，来管理Processor消费的位移。
- Processor 负责批处理完消息后，获得消息的最大位移，并更新offsets数组
- ConsumerHandler 根据 offsets，位移提交后会清空offsets集合。
- ConsumerHandler设置重平衡监听
public class Main {
public static void main(String[] args) {
String brokerList = "localhost:9092,localhost:9093,localhost:9094";
String groupId = "group2";
String topic = "test-topic";
int workerNum = 5;
ConsumerHandler consumers = new ConsumerHandler(brokerList, groupId, topic);
consumers.execute(workerNum);
try {
Thread.sleep(1000000);
} catch (InterruptedException ignored) {}
consumers.shutdown();
}
}
复制代码
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ConsumerHandler {
private final KafkaConsumer<String, String> consumer;
private ExecutorService executors;
public ConsumerHandler(String brokerList, String groupId, String topic) {
Properties props = new Properties();
props.put("bootstrap.servers", brokerList);
props.put("group.id", groupId);
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList(topic));
}
public void execute(int workerNum) {
executors = new ThreadPoolExecutor(workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());
while (true) {
ConsumerRecords<String, String> records = consumer.poll(200);
for (final ConsumerRecord record : records) {
executors.submit(new Processor(record));
}
}
}
public void shutdown() {
if (consumer != null) {
consumer.close();
}
if (executors != null) {
executors.shutdown();
}
try {
if (!executors.awaitTermination(10, TimeUnit.SECONDS)) {
System.out.println("Timeout.... Ignore for this case");
}
} catch (InterruptedException ignored) {
System.out.println("Other thread interrupted this shutdown, ignore for this case.");
Thread.currentThread().interrupt();
}
}
}
复制代码
import org.apache.kafka.clients.consumer.ConsumerRecord;
public class Processor implements Runnable {
private ConsumerRecord<String, String> consumerRecord;
public Processor(ConsumerRecord record) {
this.consumerRecord = record;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " consumed " + consumerRecord.partition()
+ "th message with offset: " + consumerRecord.offset());
}
}
复制代码
3.3 方案对比
- 第一种方案：建议采用 Consumer 单线程封装，实现多个消费者来消费（浪费资源），这样能很好地保证分区内消费的顺序，同时也没有线程切换的开销。
- 第二种方案：实现复杂，问题在于可能没法维护分区内的消息顺序，注意消息处理和消息接收解耦了。
4 Consumer指定分区消费案例实战（Standalone Consumer）
-
Standalone Consumer assign 用于接收指定分区列表的消息和Subscribe是矛盾的。只能二选一。
-
多个 Consumer 实例消费一个 Topic 借助于 group reblance可谓是天做之合。
-
若要精准控制，assign逃不了。
poperties props = new Properties(); props.put("bootstrap.servers", brokerList); props.put("group.id", groupId); props.put("enable.auto.commit", "false"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); consumer = new KafkaConsumer<>(props); List<TopicPartion> partitions = new ArrayList<>(); List<PartitionInfo> allPartitions = consumer.partitionsFor("kaiXinTopic") if(allPartitions != null && !allPartitions.isEmpty){ for(PartitionInfo partitionInfo : allPartitions){ partitions.add(new TopicPartition(partitionInfo.topic(),partitionInfo.partition())) } consumer.assign(partitions) } while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) { buffer.add(record); } if (buffer.size() >= minBatchSize) { insertIntoDb(buffer); consumer.commitSync(); buffer.clear(); } 复制代码
结语
本文综合了多本Kafka实战书籍和博客，为了写好本文，参考了大量资料，进行了语言的重组，辛苦成文，各自珍惜！
秦凯新 2181119 2123
相关文章
- 1. kafka消费者Consumer参数设置及参数调优建议-kafka 商业环境实战
- 2. kafka生产者和消费者吞吐量测试-kafka 商业环境实战
- 3. kafka生产者和消费者吞吐量测试-kafka商业环境实战
- 4. kafka生产环境规划-kafka 商业环境实战
- 5. Kafka生产与消费实战
- 6. 【Kafka】实战（二）Kafka使用 Producer 发送消息及 Consumer 接收消息消费
- 7. Kafka Consumer多线程实例
- 8. Kafka Consumer多线程消费实战和源码分析
- 9. Kafka消费组(consumer group)
- 10. Kafka消费组(consumer group)(转）
- 更多相关文章...
- • PHP 实例 - AJAX 与 XML - PHP教程
- • XML DOM 实例 - XML DOM 教程
- • Docker容器实战(六) - 容器的隔离与限制
- • Java Agent入门实战（一）-Instrumentation介绍与使用
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. kafka消费者Consumer参数设置及参数调优建议-kafka 商业环境实战
- 2. kafka生产者和消费者吞吐量测试-kafka 商业环境实战
- 3. kafka生产者和消费者吞吐量测试-kafka商业环境实战
- 4. kafka生产环境规划-kafka 商业环境实战
- 5. Kafka生产与消费实战
- 6. 【Kafka】实战（二）Kafka使用 Producer 发送消息及 Consumer 接收消息消费
- 7. Kafka Consumer多线程实例
- 8. Kafka Consumer多线程消费实战和源码分析
- 9. Kafka消费组(consumer group)
- 10. Kafka消费组(consumer group)(转）