kafka 指定partition两种方式&Consumer不消费消息的几个原因

需求

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.不使用多线程，被同一个消费者消费



2.传入key值为空，不分区，被同一个消费者消费



3.正确结果，使用多线程，key不为空