kafka消费者

1.kafka消费组基本概念

kafka消费topic是以group为单位来的，一个group消费一个topic。一个group能容纳多个consumer。consumer消费是以分区(partition)来的，一个consumer可以消费一个或多个partition，一个partition只能被一个consumer消费。(如果一个consumer group中的consumer个数多于topic中的partition的个数，多出来的consumer会闲置(idle),所以如果为了增加消费者能力，只简单增加消费者数量不一定会有用).

消费与分区对应关系

消费者数量小于partition的数量

消费者数量小于partition的数量

消费者数量小于partition的数量

2. consumer group的分区再平衡

每个consumer负责自己对应的分区，但是当group中有consumer退出或者新加入consumer，再或者topic中新增partition，group中的消费者负责的partition都得重新计算，Rebalance 期间consumer不能再消费消息，做rebalance的时候是会影响整个consumer group。

consumer获知自己消费的分区以及group内其他成员信息都是通过向一个叫做Group Coordinator的broker发送心跳来的，不同的group的broker可能不同。只要consumer再给Coordinator发送心跳，就被认为是正常的。触发心跳是通过consumer客户端轮询处理消息来的。如果consumer长时间没有心跳group coordinator就会认为consumer已经挂了，触发rebalance，新版本的java api(kafka_2.11的0.10.2.0已经支持了)支持显示的关闭客户端，这样可以避免有group coordinator因为超时来触发rebalance有此导致消息积压。

  分区分配流程:
  1.第一个加入group的consumer是consumer的leader（这个consumer奔溃之后会怎么样暂时不清楚）
  2.新加入的consumer向group coordinator发送加入请求
  3.leader从group coordinator接收消费者列表，然后给每个consumer分配分区
  4.leader将重新分配的信息发送给group coordinator，group coordinator再将信息发送给所有的consumer

3.启动一个consumer

使用java api只需要配置 bootstrap.servers, key.deserializer, value.deserializer三个配置就可以。一般还要带上group.id，指定所属的消费组。

  Properties properties = new Properties();
        properties.put("bootstrap.servers", "127.0.0.1:9092");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("group.id", "testTopicGroup1");
        new KafkaConsumer<String, String>(properties);

4.订阅topic

consumer.subscribe(Collections.singletonList("testTopic"));
可以指定多个topic，可以使用正则表达式：
consumer.subscribe("test.*");
demo:

 private volatile boolean shutdown = false;
    public void poll(){

        Properties properties = new Properties();
        properties.put("bootstrap.servers", "120.27.8.221:9092");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("group.id", "testGroup");
        KafkaConsumer consumer = new KafkaConsumer<String, String>(properties);

        //关闭轮询
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable(){
            @Override
            public void run() {
                shutdown = true;
            }
        }));

        try{
            while (!shutdown){
                //开始轮询消息，poll会找到group coordinator，加入consumer group，确认消费的分区，获取消息
                //poll会获取本地最大的offset之后的消息，而不是commit到kafka中的offset
                ConsumerRecords<String, String> records = consumer.poll(100);
                for(ConsumerRecord record:records){
                    System.out.println("topic = " + record.topic()
                      + ", partition = " + record.partition()
                      + ", offset = " + record.offset()
                      + ", customer = " + record.key()
                      + ", country = " + record.value());
                }
            }
        }finally {
            //及时关闭消费者
            consumer.close();
        }

    }

5.commit offset

不管什么时候调用poll方法都会获取到还未被消费过的消息，这个实现通过消息的offset来实现的，每个分区的offset的管理是通过consumer自己向一个特殊的topic（__consumer_offsets）提交消息来实现的.

1.自动提交

开启自动提交之后，在每次调用poll获取消息的时候会检查时间查看是否需要提交offset，如果已经到时间之后会提交offset，自动提交的好处是方便，劣势是不能灵活控制，如果间隔期间consumer奔溃，已经处理且未提交的消息会被处理两遍。
自动提交配置：
enable.auto.commit=true ##开启自动提交，默认5s提交一次
auto.commit.interval.ms=1000 ##设置自动提交时间间隔