设为首页 加入收藏

TOP

kafka 指定partition两种方式&Consumer不消费消息的几个原因
2019-03-25 14:33:37 】 浏览:65
Tags:kafka 指定 partition 方式 &Consumer 消费 消息 原因
版权声明:本文为博主原创文章,未经博主允许不得转载。 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) {
        // TODO Auto-generated method stub
        /*
         * Cluster arg5打印出来是这个样子
         * Cluster(nodes = [172.17.11.11:9092 (id: 0 rack: null), 172.17.11.13:9092 (id: 1 rack: null), 172.17.11.15:9092 (id: 2 rack: null)], partitions = [Partition(topic = TOPIC-20160504-1200, partition = 1, leader = 2, replicas = [0,1,2,], isr = [2,1,0,], Partition(topic = TOPIC-20160504-1200, partition = 2, leader = 0, replicas = [0,1,2,], isr = [0,2,1,], Partition(topic = TOPIC-20160504-1200, partition = 0, leader = 1, replicas = [0,1,2,], isr = [1,0,2,]])
        */
        /*
         * byte[] arg2以字节码的格式存储key
         * System.out.println(new String(arg2));
         * System.out.println(key.toString());二者输出相同,都是key
         * byte[] arg4和Object value同理
         */
        /*
         * 返回值指定的分区值
         */
         //从传入的key中分割出用于分区的数值
        int partition= Integer.parseInt(key.toString().split("\\.")[3]);//分割 " . " 需要转义" \\. "
        if(partition%2==0){
            return 1;
        }else{
            return 2;
        }
    }

完整代码,可以参考我的github

Consumer不消费消息的几个原因

通过打印当前对象地址,来确定两个分区是否被两个不同的消费者消费,发现即使我使用了两个消费者,实际消费消息的还是一个消费者,只是会消费完一个分区再去消费第二个分区,第二个消费者并没有起到作用,需要多线程
1.不使用多线程,被同一个消费者消费
clipboard.png
2.传入key值为空,不分区,被同一个消费者消费
clipboard1.png
3.正确结果,使用多线程,key不为空
clipboard3.png

】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇Error while fetching metadata w.. 下一篇Flume (九) Channels - Kafka Cha..

最新文章

热门文章

Hot 文章

Python

C 语言

C++基础

大数据基础

linux编程基础

C/C++面试题目