kafka的消费者从属于消费者群组，一个群组里的消费者订阅的是同一个主题，每个消费者接受主题一部分分区的消息。

分区的所有权从一个消费者转移到另一个消费者，这样的行为被称为再均衡。

消费者通过向被指派为群组协调器的broker（不同的群组可以有不同的协调器）发送心跳来维持它们和群组的从属关系以及它们对分区的所有权关系。

分配分区，它使用了一个实现了PartitionAssigor接口的类来决定哪些分区应该被分配给哪个消费者。

创建kafka消费者

1. 使用3个必要的属性：bootstrap.servers,key.deserializer和value.deserializer

• 第一个属性bootstrap.servers指定了kafka集群的连接字符串。它的用途在kafkaProducer中的用途是一样的。
• 另外两个属性value.deserializer和key.deserializer，使用指定的类把字节数组转换成java对象。
• group.id指定了kafkaCounsumer属于哪一个消费者群组。

//创建一个kafkaConsumer对象
Properties props = new Properties();
props.put("bootstrap.servers","broker1:9092,broker2:9092");
props.put("group.id","CountryCounter");
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<String,String>(props);

2. 订阅主题subscribe()方法接受一个主题列表作为参数

 consumer.subscribe(Collection.singletonList("costomerCountries"));

可以传入一个正则表达式，可以匹配多个主题，如果有新的主题创建，并且主题的名字与正则表达式匹配，那么会立即触发一次在均衡，消费者就可以读取新添加的主题。例如要订阅所有与test相关的主题，可以这样做：

consumer.subscribe("test.*");

轮询

消息轮询是消费者API 的核心，一旦消费者订阅了主题，轮询就会处理所有细节，包括群组协调、分区再均衡。发送心跳和获取数据。

try{
    while(true){
        ConsumerRecords<String,String> records = consumer.poll(100);
        for(ConsumerRecord<String,String> record : records){
            log.debug("topic=%s,partition=%s,offse=%d,customer=%s,country=%s\n",record.topic,record.partitions(),record,offset(),record.key(),record.value());
            
            int updatedCounty =1;
            if(custCountryMap.countainsValue(record.value())){
                updateCounty = custCountryMap.get(record.value())+1;
            }
            custCountryMap.put(record.value(),updatedCounty)
            
            JSONObject json = new JsonObject(custCountyMap);
            System.out.println(json.toString(4));
            
        }
    }
}finally{
    consumer.close();
}

提交和偏移量

我们把更新分区当前位置的操作叫做提交。

1. 自动提交
2. 提交当前偏移量 把enable.auto.commit 设置为false 让应用程序决定何时提交偏移量。使用commitSync()提交偏移量是最简单可靠的。这个API会提交由poll()方法返回最新的偏移量，提交成功后马上返回，如果提交失败就抛出异常。

//使用commitSync()方法提交偏移量的例子
while(true){
    ConsumerRecords<String,String> records = consumer.poll(100);
    for(ConsumerRecord<String,String> record : records){
        //此处处理提取后的数据
    }
    try{
        consumer.commitSync();
    }catch(CommitFailedException e){
        log.error("commit failed",e)
    }
}

3.异步提交

while(true){
    ConsumerRecords<String,String> records = consumer.poll(100);
    for(ConsumerRecord<String,String> record : records){
        //此处处理提取后的数据
        
    }
    consumer.commitAsync();
}

4.提交特定的偏移量
调用commitSYNC()时提交分区和偏移量的map

parivate Map<TopicPartition,OffserAndMetadata> currentOffsets = new HashMap<>();
int count = 0;

while(true){
    ConsumerRecords<String,String> records = consumer.poll(100);
    for(ConsumerRecord<String,String> record : records){
        //此处处理提取后的数据
        currentOffsets.put(new TopicPartition(record.topic(),record.partition()),new OffsetAndMetadata(record.offset()+1,"no metadata"));
        if(count%1000==0){
            consumer.commitAsync(currentOffsets,null);
            count++:
        }
    }
}

再均衡监听器

在为消费者分配新分区或移除旧分区时，可以通过消费者API执行一些应用程序代码，在调用subscribe()方法是传进去一个ConsumerRebalanceListener实例就可以了。ConsumerRebalanceListener有两个需要实现的方法。

• public void onPartitionsRevoked(Collection partitions)方法会在再均衡开始之前和消费者停止读取消息之后被调用，如果在这里提交偏移量，下一个接管分区的消费者就知道该从哪里开始读取了。
• public void onPartitionsAssigned(Collection partitions)方法会在重新分配分区之后和消费者开始读取消息之前调用

private Map<TopicPartition,OffsetAndMetadata> currentOffsets = new HashMap<>();

private class HandleRebalace implements ConsumerRebalanceListener{
    public void onPartitionsAssigned(Collection<TopicPartition> partitions){
        //重新分配分区之后和消费者开始读取消息之前
    }
    
    public void onPartitionsRevoked(Collection<TopicPartition> partitions){
        //再均衡之前和消费者停止读取消息之后
        consumer.commitSync(currentOffsets);
    }
    
    try{
        consumer.subscribe(topics,new HandleRebalance());
        
        while(true){
            ConsumerRecords<String,String> records = consumer.poll(100);
            for(ConsumerRecord<String,String> record : records){
                //此处处理提取后的数据
                currentOffsets.put(new TopicPartition(record.topic(),record.partition()),new OffsetAndMetadata(record.offset()+1,"no metadata"));
            }
            consumer.commitAsync(currentOffsets,null);
        }
    }catch(Exception e){
        e.printException();
    }finally{
        try{
            consumer.commitSync(currentOffsets);
        }finally{
            consumer.close();
        }
    }
}

独立消费者

不需要消费者群组和再均衡

List<PartitionInfo> partitionInfos = null;
partitionInfos = consumer.partitionsFor("topic");

if(partitionInfos !=null){
    for(PartitionInfo partition : partitionInfos){
        partitions.add(new TopicPartition(partition.topic(),partition.partition()));
    }
    consumer.assign(partitions);
    
}

如果主题增加了新的分区，消费者并不会收到通知。所以，要么周期性的调用consumer.partitionsFor()方法来检查是否有新分区加入，要么在添加新分区后重启应用程序。