依赖

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.3.0</version>
</dependency>

介绍

本示例中，生产者发送50条消息给有3个消费者的群组。消费者群组中，第三个线程会中途退出群组，借此，我们可以观察分区再均衡现象。

代码

生产者

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class RebalanceProducer {private static final int MSG_SIZE = 50;private static ExecutorService executorService= Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());private static CountDownLatch countDownLatch= new CountDownLatch(MSG_SIZE);public static void main(String[] args) {Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.100.14:9092");properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);KafkaProducer<String,String> producer = new KafkaProducer(properties);try {for(int i=0;i<MSG_SIZE;i++){ProducerRecord<String,String> record= new ProducerRecord("rebalance-topic-three-part","value" + i);executorService.submit(new ProduceWorker(record,producer,countDownLatch));Thread.sleep(600);}countDownLatch.await();} catch (Exception e) {e.printStackTrace();} finally {producer.close();executorService.shutdown();}}
}

生产任务

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;import java.util.concurrent.CountDownLatch;public class ProduceWorker implements Runnable{private ProducerRecord<String,String> record;private KafkaProducer<String,String> producer;private  CountDownLatch countDownLatch;public ProduceWorker(ProducerRecord<String, String> record,KafkaProducer<String, String> producer, CountDownLatch countDownLatch) {this.record = record;this.producer = producer;this.countDownLatch = countDownLatch;}public void run() {final String id = "" + Thread.currentThread().getId();try {producer.send(record, new Callback() {public void onCompletion(RecordMetadata metadata,Exception exception) {if(null!=exception){exception.printStackTrace();}if(null!=metadata){System.out.println(id+"|"+String.format("偏移量：%s,分区：%s",metadata.offset(),metadata.partition()));}}});System.out.println(id+":数据["+record.key()+ "-" + record.value()+"]已发送。");countDownLatch.countDown();} catch (Exception e) {e.printStackTrace();}}
}

消费者

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class RebalanceConsumer {public static final String GROUP_ID = "RebalanceConsumer";private static ExecutorService executorService= Executors.newFixedThreadPool(3);public static void main(String[] args) throws InterruptedException {Properties properties = new Properties();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.100.14:9092");properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);properties.put(ConsumerConfig.GROUP_ID_CONFIG, RebalanceConsumer.GROUP_ID);properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);for(int i = 0; i < 2; i++){executorService.submit(new ConsumerWorker(false, properties));}Thread.sleep(5000);//用来被停止，观察保持运行的消费者情况new Thread(new ConsumerWorker(true, properties)).start();}
}

消费任务

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;public class ConsumerWorker  implements Runnable{private final KafkaConsumer<String,String> consumer;/*用来保存每个消费者当前读取分区的偏移量*/private final Map<TopicPartition, OffsetAndMetadata> currOffsets;private final boolean isStop;/*消息消费者配置*/public ConsumerWorker(boolean isStop, Properties properties) {this.isStop = isStop;this.consumer= new KafkaConsumer(properties);this.currOffsets= new HashMap();consumer.subscribe(Collections.singletonList("rebalance-topic-three-part"),new HandlerRebalance(currOffsets,consumer));}public void run() {final String id = "" + Thread.currentThread().getId();int count = 0;TopicPartition topicPartition;long offset;try {while(true){ConsumerRecords<String, String> records= consumer.poll(Duration.ofMillis(500));//业务处理//开始事务for(ConsumerRecord<String, String> record:records){System.out.println(id+"|"+String.format("处理主题：%s，分区：%d，偏移量：%d，" +"key：%s，value：%s",record.topic(),record.partition(),record.offset(),record.key(),record.value()));topicPartition = new TopicPartition(record.topic(),record.partition());offset = record.offset()+1;currOffsets.put(topicPartition,new OffsetAndMetadata(offset,"no"));count++;//执行业务sql}if(currOffsets.size()>0){for(TopicPartition topicPartitionkey:currOffsets.keySet()){HandlerRebalance.partitionOffsetMap.put(topicPartitionkey,currOffsets.get(topicPartitionkey).offset());}//提交事务,同时将业务和偏移量入库}//如果stop参数为true,这个消费者消费到第5个时自动关闭if(isStop&&count>=5){System.out.println(id+"-将关闭，当前偏移量为："+currOffsets);consumer.commitSync();break;}consumer.commitSync();}} finally {consumer.close();}}
}

再均衡监听器

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;public class HandlerRebalance implements ConsumerRebalanceListener {private final Map<TopicPartition, OffsetAndMetadata> currOffsets;private final KafkaConsumer<String,String> consumer;//private final Transaction  tr事务类的实例public HandlerRebalance(Map<TopicPartition, OffsetAndMetadata> currOffsets,KafkaConsumer<String, String> consumer) {this.currOffsets = currOffsets;this.consumer = consumer;}/*模拟一个保存分区偏移量的数据库表*/public final static ConcurrentHashMap<TopicPartition,Long>partitionOffsetMap = new ConcurrentHashMap();//分区再均衡之前public void onPartitionsRevoked(Collection<TopicPartition> partitions) {final String id = Thread.currentThread().getId()+"";System.out.println(id+"-onPartitionsRevoked参数值为："+partitions);System.out.println(id+"-服务器准备分区再均衡，提交偏移量。当前偏移量为："+currOffsets);//开始事务//偏移量写入数据库System.out.println("分区偏移量表中："+partitionOffsetMap);for(TopicPartition topicPartition:partitions){partitionOffsetMap.put(topicPartition,currOffsets.get(topicPartition).offset());}consumer.commitSync(currOffsets);//提交业务数和偏移量入库  tr.commit}//分区再均衡完成以后public void onPartitionsAssigned(Collection<TopicPartition> partitions) {final String id = "" + Thread.currentThread().getId();System.out.println(id+"-再均衡完成，onPartitionsAssigned参数值为："+partitions);System.out.println("分区偏移量表中："+partitionOffsetMap);for(TopicPartition topicPartition:partitions){System.out.println(id+"-topicPartition"+topicPartition);//模拟从数据库中取得上次的偏移量Long offset = partitionOffsetMap.get(topicPartition);if(offset==null) continue;//从特定偏移量处开始记录 (从指定分区中的指定偏移量开始消费)//这样就可以确保分区再均衡中的数据不错乱consumer.seek(topicPartition,partitionOffsetMap.get(topicPartition));}}
}

结果

生产者

14:数据[null-value0]已发送。
14|偏移量：51,分区：0
15:数据[null-value1]已发送。
15|偏移量：51,分区：2
16:数据[null-value2]已发送。
16|偏移量：49,分区：1
17:数据[null-value3]已发送。
17|偏移量：52,分区：0
18:数据[null-value4]已发送。
18|偏移量：52,分区：2
19:数据[null-value5]已发送。
19|偏移量：50,分区：1
20:数据[null-value6]已发送。
20|偏移量：53,分区：0
21:数据[null-value7]已发送。
21|偏移量：53,分区：2
14:数据[null-value8]已发送。
14|偏移量：51,分区：1
15:数据[null-value9]已发送。
15|偏移量：54,分区：0
16:数据[null-value10]已发送。
16|偏移量：54,分区：2
17:数据[null-value11]已发送。
17|偏移量：52,分区：1
18:数据[null-value12]已发送。
19:数据[null-value13]已发送。
20:数据[null-value14]已发送。
18|偏移量：55,分区：0
21:数据[null-value15]已发送。
14:数据[null-value16]已发送。
19|偏移量：55,分区：2
20|偏移量：53,分区：1
21|偏移量：56,分区：0
14|偏移量：56,分区：2
15:数据[null-value17]已发送。
15|偏移量：54,分区：1
16:数据[null-value18]已发送。
16|偏移量：57,分区：0
17:数据[null-value19]已发送。
17|偏移量：57,分区：2
18:数据[null-value20]已发送。
19:数据[null-value21]已发送。
18|偏移量：55,分区：1
19|偏移量：58,分区：0
20:数据[null-value22]已发送。
20|偏移量：58,分区：2
21:数据[null-value23]已发送。
21|偏移量：56,分区：1
14:数据[null-value24]已发送。
14|偏移量：59,分区：0
15:数据[null-value25]已发送。
15|偏移量：59,分区：2
16:数据[null-value26]已发送。
16|偏移量：57,分区：1
17:数据[null-value27]已发送。
17|偏移量：60,分区：0
18:数据[null-value28]已发送。
18|偏移量：60,分区：2
19:数据[null-value29]已发送。
19|偏移量：58,分区：1
20:数据[null-value30]已发送。
20|偏移量：61,分区：0
21:数据[null-value31]已发送。
21|偏移量：61,分区：2
14:数据[null-value32]已发送。
14|偏移量：59,分区：1
15:数据[null-value33]已发送。
15|偏移量：62,分区：0
16:数据[null-value34]已发送。
16|偏移量：62,分区：2
17:数据[null-value35]已发送。
17|偏移量：60,分区：1
18:数据[null-value36]已发送。
18|偏移量：63,分区：0
19:数据[null-value37]已发送。
19|偏移量：63,分区：2
20:数据[null-value38]已发送。
20|偏移量：61,分区：1
21:数据[null-value39]已发送。
21|偏移量：64,分区：0
14:数据[null-value40]已发送。
14|偏移量：64,分区：2
15:数据[null-value41]已发送。
15|偏移量：62,分区：1
16:数据[null-value42]已发送。
16|偏移量：65,分区：0
17:数据[null-value43]已发送。
17|偏移量：65,分区：2
18:数据[null-value44]已发送。
18|偏移量：63,分区：1
19:数据[null-value45]已发送。
19|偏移量：66,分区：0
20:数据[null-value46]已发送。
20|偏移量：66,分区：2
21:数据[null-value47]已发送。
21|偏移量：64,分区：1
14:数据[null-value48]已发送。
14|偏移量：67,分区：0
15:数据[null-value49]已发送。
15|偏移量：67,分区：2

消费者

启动生产者之前

13-onPartitionsRevoked参数值为：[]
14-onPartitionsRevoked参数值为：[]
13-服务器准备分区再均衡，提交偏移量。当前偏移量为：{}
14-服务器准备分区再均衡，提交偏移量。当前偏移量为：{}
分区偏移量表中：{}
分区偏移量表中：{}
17-onPartitionsRevoked参数值为：[]
17-服务器准备分区再均衡，提交偏移量。当前偏移量为：{}
分区偏移量表中：{}
14-再均衡完成，onPartitionsAssigned参数值为：[rebalance-topic-three-part-1]
分区偏移量表中：{}
14-topicPartitionrebalance-topic-three-part-1
17-再均衡完成，onPartitionsAssigned参数值为：[rebalance-topic-three-part-2]
分区偏移量表中：{}
13-再均衡完成，onPartitionsAssigned参数值为：[rebalance-topic-three-part-0]
分区偏移量表中：{}
13-topicPartitionrebalance-topic-three-part-0
17-topicPartitionrebalance-topic-three-part-2

线程13 —— 分区0
线程14 —— 分区1
线程17 —— 分区2

启动生产者后，第三线程关闭之前

13|处理主题：rebalance-topic-three-part，分区：0，偏移量：51，key：null，value：value0
17|处理主题：rebalance-topic-three-part，分区：2，偏移量：51，key：null，value：value1
14|处理主题：rebalance-topic-three-part，分区：1，偏移量：49，key：null，value：value2
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：52，key：null，value：value3
17|处理主题：rebalance-topic-three-part，分区：2，偏移量：52，key：null，value：value4
14|处理主题：rebalance-topic-three-part，分区：1，偏移量：50，key：null，value：value5
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：53，key：null，value：value6
17|处理主题：rebalance-topic-three-part，分区：2，偏移量：53，key：null，value：value7
14|处理主题：rebalance-topic-three-part，分区：1，偏移量：51，key：null，value：value8
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：54，key：null，value：value9
17|处理主题：rebalance-topic-three-part，分区：2，偏移量：54，key：null，value：value10
14|处理主题：rebalance-topic-three-part，分区：1，偏移量：52，key：null，value：value11
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：55，key：null，value：value12
17|处理主题：rebalance-topic-three-part，分区：2，偏移量：55，key：null，value：value13
//注意这里，上面17处理的偏移量是55，处理完后，偏移量到了56，如下
17-将关闭，当前偏移量为：{rebalance-topic-three-part-2=OffsetAndMetadata{offset=56, leaderEpoch=null, metadata='no'}}
14|处理主题：rebalance-topic-three-part，分区：1，偏移量：53，key：null，value：value14
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：56，key：null，value：value15
//同17，处理完后，偏移量是57
14|处理主题：rebalance-topic-three-part，分区：1，偏移量：54，key：null，value：value17
//处理完后，偏移量是58
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：57，key：null，value：value18

线程13 —— 分区0 —— 偏移量 58
线程14 —— 分区1 —— 偏移量 55
线程17 —— 分区2 —— 偏移量 56

第三线程关闭后，分区再平衡

13-onPartitionsRevoked参数值为：[rebalance-topic-three-part-0]
14-onPartitionsRevoked参数值为：[rebalance-topic-three-part-1]
13-服务器准备分区再均衡，提交偏移量。当前偏移量为：{rebalance-topic-three-part-0=OffsetAndMetadata{offset=58, leaderEpoch=null, metadata='no'}}
分区偏移量表中：{rebalance-topic-three-part-2=56, rebalance-topic-three-part-1=55, rebalance-topic-three-part-0=58}
14-服务器准备分区再均衡，提交偏移量。当前偏移量为：{rebalance-topic-three-part-1=OffsetAndMetadata{offset=55, leaderEpoch=null, metadata='no'}}
分区偏移量表中：{rebalance-topic-three-part-2=56, rebalance-topic-three-part-1=55, rebalance-topic-three-part-0=58}
14-再均衡完成，onPartitionsAssigned参数值为：[rebalance-topic-three-part-2]
13-再均衡完成，onPartitionsAssigned参数值为：[rebalance-topic-three-part-1, rebalance-topic-three-part-0]
分区偏移量表中：{rebalance-topic-three-part-2=56, rebalance-topic-three-part-1=55, rebalance-topic-three-part-0=58}
13-topicPartitionrebalance-topic-three-part-1
分区偏移量表中：{rebalance-topic-three-part-2=56, rebalance-topic-three-part-1=55, rebalance-topic-three-part-0=58}
14-topicPartitionrebalance-topic-three-part-2
13-topicPartitionrebalance-topic-three-part-0

线程13 —— 分区0 —— 偏移量 58 —— 分区1 —— 偏移量 —— 55
线程14 —— 分区2 —— 偏移量 56

分区再平衡之后

线程13 —— 分区0 —— 偏移量 58 —— 分区1 —— 偏移量 —— 55
线程14 —— 分区2 —— 偏移量 56

14|处理主题：rebalance-topic-three-part，分区：2，偏移量：56，key：null，value：value16
14|处理主题：rebalance-topic-three-part，分区：2，偏移量：57，key：null，value：value19
13|处理主题：rebalance-topic-three-part，分区：1，偏移量：55，key：null，value：value20
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：58，key：null，value：value21
14|处理主题：rebalance-topic-three-part，分区：2，偏移量：58，key：null，value：value22
13|处理主题：rebalance-topic-three-part，分区：1，偏移量：56，key：null，value：value23
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：59，key：null，value：value24
14|处理主题：rebalance-topic-three-part，分区：2，偏移量：59，key：null，value：value25
13|处理主题：rebalance-topic-three-part，分区：1，偏移量：57，key：null，value：value26
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：60，key：null，value：value27
14|处理主题：rebalance-topic-three-part，分区：2，偏移量：60，key：null，value：value28
13|处理主题：rebalance-topic-three-part，分区：1，偏移量：58，key：null，value：value29
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：61，key：null，value：value30
14|处理主题：rebalance-topic-three-part，分区：2，偏移量：61，key：null，value：value31
13|处理主题：rebalance-topic-three-part，分区：1，偏移量：59，key：null，value：value32
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：62，key：null，value：value33
14|处理主题：rebalance-topic-three-part，分区：2，偏移量：62，key：null，value：value34
13|处理主题：rebalance-topic-three-part，分区：1，偏移量：60，key：null，value：value35
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：63，key：null，value：value36
14|处理主题：rebalance-topic-three-part，分区：2，偏移量：63，key：null，value：value37
13|处理主题：rebalance-topic-three-part，分区：1，偏移量：61，key：null，value：value38
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：64，key：null，value：value39
14|处理主题：rebalance-topic-three-part，分区：2，偏移量：64，key：null，value：value40
13|处理主题：rebalance-topic-three-part，分区：1，偏移量：62，key：null，value：value41
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：65，key：null，value：value42
14|处理主题：rebalance-topic-three-part，分区：2，偏移量：65，key：null，value：value43
13|处理主题：rebalance-topic-three-part，分区：1，偏移量：63，key：null，value：value44
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：66，key：null，value：value45
14|处理主题：rebalance-topic-three-part，分区：2，偏移量：66，key：null，value：value46
13|处理主题：rebalance-topic-three-part，分区：1，偏移量：64，key：null，value：value47
13|处理主题：rebalance-topic-three-part，分区：0，偏移量：67，key：null，value：value48
14|处理主题：rebalance-topic-three-part，分区：2，偏移量：67，key：null，value：value49

参考：King——笔记-Kafka

Kafka 再均衡监听器示例相关推荐

kafka再均衡监听器测试
[README] 本文使用的kafka是最新的 kafka3.0.0:本文kafka集群有3个节点分别是 centos201, centos202, centos203 : brokerid 分别为 ...
【Kafka】kafka 再均衡监听器 ConsumerRebalanceListener
1.概述新版本 consumer 默认把位移提交到 consumer offsets 中 .其Kafka 也支持用户把位移提交到外部存储中,比如数据库中.若要实现这个功能,用户就必须使用 rebal ...
关于Kafka中的再均衡
本文来说下Kafka中的再均衡文章目录概述触发时机协调者交互方式处理流程本文小结概述在Kafka消费者的使用和原理中已经提到过"再均衡"的概念,我们先回顾下,一个 ...
学Kafka，就必须了解的再均衡问题！
作者 | 草捏子来源 | 草捏子(ID:chaycao) 头图 | CSDN 下载自东方IC 在<Kafka消费者的使用和原理>中已经提到过"再均衡"的概念,我们先 ...
Kafka | Kafka的消费再均衡是指什么？
记录收藏一波强大网友的解读: https://www.cnblogs.com/luozhiyun/p/11909315.html 摘自原文: 就目前而言,一共有如下几种情形会触发再均衡的操作: 有新的 ...
Kafka负载均衡策略
分区器分区器是生产者层面的负载均衡.Kafka 生产者生产消息时,根据分区器将消息投递到指定的分区中,所以 Kafka 的负载均衡很大程度上依赖于分区器. Kafka 默认的分区器是 Kafka 提 ...
华为MSTP负载均衡配置示例
华为MSTP负载均衡配置示例 :https://www.cnblogs.com/zhuimengle/p/5906806.html MSTP负载均衡配置示例本示例拓扑结构如图8-38所示,Switc ...
Kafka JAVA客户端代码示例--高级应用
2019独角兽企业重金招聘Python工程师标准>>> 什么时间使用高级应用? 针对一个消息读取多次在一个process中,仅仅处理一个topic中的一组partitions 使用 ...
windows环境安装haproxy及初步配置负载均衡使用示例
安装HaProxy 首先需要下载windows环境下需要文件,这里下载的是一个别人编译好的一个文件,这里省去了编译的过程,使用的版本是haproxy-1.7.8. 下载后直接解压到对应的目录下.示例( ...

Kafka 再均衡监听器示例