版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/Nougats/article/details/71602660
需求
1.随机生成IP数字,将奇数、偶数数据分在不同的Partition
2.通过KafkaAPI实现一个消费组中多消费者,为了是验证同组的不同消费者是否一一对应不同的Patition
3.打包部署在Linux中运行
解决办法
方法1:
在producer发送消息时指定partition,ProducerRecord的构造方法可以有四个参数,分别是topic,int类型的partition值,key,value,我们直接指定传入的第二个参数即可
当我们不指定第二个参数,使用三个参数的构造方式时,会根据传入的key自动分区,传入key为空时消息不分区,会传到同一个partition中
producer.send(new ProducerRecord<>(topic,partition,ip, msg));
方法2:
自定义Partitioner,重写partition方法
@Override
public int partition(String topic, Object key, byte[] arg2, Object value,
byte[] arg4, Cluster arg5) {
int partition= Integer.parseInt(key.toString().split("\\.")[3]);
if(partition%2==0){
return 1;
}else{
return 2;
}
}
完整代码,可以参考我的github
Consumer不消费消息的几个原因
通过打印当前对象地址,来确定两个分区是否被两个不同的消费者消费,发现即使我使用了两个消费者,实际消费消息的还是一个消费者,只是会消费完一个分区再去消费第二个分区,第二个消费者并没有起到作用,需要多线程
1.不使用多线程,被同一个消费者消费
2.传入key值为空,不分区,被同一个消费者消费
3.正确结果,使用多线程,key不为空