消费者与组

Kafka 消费者从属于消费者群组。消费者群组内的消费者订阅的是相同的主题，每个消费者接收主题一部分分区的消息。

1. 主题 T1 有4个分区，仅存在消费者 C1， C1 将收到全部4个分区的消息

 2. 在群组 G1 里新增消费者 C2，那么每个消费者将分别从两个分区接收消息。

 3. 如果群组 G1 有 4 个消费者，那么每个消费者可以分配到一个分区

 4. 如果我们往群组里添加更多的消费者，超过主题的分区数量，那么有一部分消费者就会被闲置，不会接收到任何消息

 

5. 如果新增一个只包含一个消费者的群组 G2，那么这个消费者将从主题 T1 上接收所有的消息，与群组 G1 之间互不影响。

简而言之，为每一个需要获取一个或多个主题全部消息的应用程序创建一个消费者群组， 然后往群组里添加消费者来伸缩读取能力和处理能力，群组里的每个消费者只处理一部分消息。

 

核心配置

1. bootstrap.servers （从哪里获取）

2. group.id（属于哪个群组）

3. key.deserializer(键反序列化器)

4. value.deserializer(值反序列化器)

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);

订阅主题

consumer.subscribe(Collections.singletonList("customerCountries"));

支持准确的主题，也支持正则。

consumer.subscribe("test.*");

轮询消息

try {
    // 轮询
    while (true) {
        // 超时100
        ConsumerRecords<String, String> records = consumer.poll(100);
        // 遍历取到的消息
        for (ConsumerRecord<String, String> record : records){
            log.debug("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());

            int updatedCount = 1;
            if (custCountryMap.countainsValue(record.value())) {
                updatedCount = custCountryMap.get(record.value()) + 1; 
            } 

            custCountryMap.put(record.value(), updatedCount)
            
            JSONObject json = new JSONObject(custCountryMap);
            System.out.println(json.toString(4));
        }
    }
} finally {
    // 关闭
    consumer.close();
}

通过poll方法拉消息，传入的100是超时时间。在退出应用程序之前使用 close() 方法关闭消费者。

几种特殊的配置

属性	说明
fetch.min.bytes	消费者从服务器获取记录的最小字节数
fetch.max.wait.ms	指定 broker 的等待时间，默认是 500ms 与fetch.min.bytes配合使用
max.partition.fetch.bytes	从每个分区里返回给消费者的最大字节数
session.timeout.ms	消费者在被认为死亡之前可以与服务器断开连接的时间，默认是 3s
heartbeat.interval.ms	poll() 方法向协调器发送心跳的频率, 一般是session.timeout.ms的1/3
auto.offset.reset	消费者在读取一个没有偏移量的分区或者偏移量无效的情况下(因消费者长 时间失效，包含偏移量的记录已经过时并被删除)该作何处理 默认值是 latest，从最新的记录开始读取数据。另一个值是 earliest，从起始位置读取分区的记录。
enable.auto.commit	消费者是否自动提交偏移 量，默认值是 true. 如果把它设为 true，还可以通过配置 auto.commit.interval.ms 属性来控制提交的频率
partition.assignment.strategy	决定哪些分区应该被分配给哪个消费者. Range 该策略会把主题的若干个连续的分区分配给消费者. RoundRobin 策略会给所 有消费者分配相同数量的分区(或最多就差一个分区)
client.id	标识从客户端发送过来的消息
max.poll.records	控制单次调用 poll() 方法能够返回的记录数量
receive.buffer.bytes send.buffer.bytes	读写数据时用到的 TCP 缓冲区大小

提交、偏移量

调用 poll() 方法会返回还没有被消费者读取过的记录， 可以追踪到哪些记录是被群组里的哪个消费者读取的。消费者可以使用 Kafka 来追踪消息在分区里的位置(偏移量)。

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

消费者往一个叫作 _consumer_offset 的主题发送消息，消息里包含每个分区的偏移量。消费者发生崩溃或者有新的消费者加入群组，会触发再均衡，分区会重新分配。为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的地方继续处理。

总之，偏移量是为了发生再平衡时，重新分配的消费者能够继续之前的位置继续处理数据，而不是丢失数据或重复消费已经处理过的数据。

而提交偏移量就是为了防止发生再平衡时，找不到这个偏移量（车祸现场）。

自动提交

最简单的提交方式是让消费者自动提交偏移量。如果 enable.auto.commit 被设为 true，那 么每过 5s，消费者会自动把从 poll() 方法接收到的最大偏移量提交上去。

但是无法避免的是在两次提交偏移量期间，发生再平衡，导致已经处理的偏移量没有及时提交，会导致重复消费。

手动同步提交

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100); 
    for (ConsumerRecord<String, String> record : records){
        System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
    }

    try {
        // 同步提交
        consumer.commitSync();
    } catch (CommitFailedException e) {
        log.error("commit failed", e)
    }

}

把 auto.commit.offset 设为 false，让应用程序决定何时提交偏移量. commitSync() 将会提交由 poll() 返回的最新偏移量。如果发生了再均衡，从最近一批消息到发生再均衡之间的所有消息都将被重复处理。commitSync在提交发生失败是会进行重试。

手动异步提交

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100); 
    for (ConsumerRecord<String, String> record : records){
        System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
    }
    consumer.commitAsync();
}

异步提交支持回调

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100); 
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
    }

    consumer.commitAsync(new OffsetCommitCallback() {
             public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
                    if (e != null)
                        log.error("Commit failed for offsets {}", offsets, e);
            } 
    });
}

commitAsync()在发生失败时不会进行重试。

组合式提交

为了避免消费者关闭时无法提交偏移量，在关闭时执行commitSync()，在每次处理时使用commitAsync()这样可以提高吞吐量的同时，保证数据的提交。

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100); 
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
        }
        // 单消息异步提交
        consumer.commitAsync();
    }
} catch (Exception e) { 
    log.error("Unexpected error", e);
} finally {
    try {
        // 退出时同步提交
        consumer.commitSync();
    } finally {
        consumer.close(); 
    }
}

分partition提交偏移量

// 分区偏移量的映射
Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
int count = 0; 

while (true) {
    // 拉消息
    ConsumerRecords<String, String> records = consumer.poll(100); 
    for (ConsumerRecord<String, String> record : records){
        System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
        // 记录偏移量
        currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset()+1, "no metadata"));
        // 每1000条提交一次
        if (count % 1000 == 0)
            consumer.commitAsync(currentOffsets, null);

        count++;
    } 
}

这里手动记录了不同topic不同partition下的偏移量，然后传递给commit方法提交。

解决再均衡时，偏移量无法及时提交的问题——再均衡监听器

// 记录偏移
Map<TopicPartition, OffsetAndMetadata> currentOffsets= new HashMap<>();

private class HandleRebalance implements ConsumerRebalanceListener { 

    public void onPartitionsAssigned(Collection<TopicPartition>partitions) {

    }
    
    // 再均衡
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        System.out.println("Lost partitions in rebalance. Committing current
offsets:" + currentOffsets);
        // 同步提交偏移
        consumer.commitSync(currentOffsets);
    }
}

try {
    // 注册监听器
    consumer.subscribe(topics, new HandleRebalance());
         
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
            currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset()+1, "no metadata"));
        }
        // 异步提交，不会重试
        consumer.commitAsync(currentOffsets, null); }
} catch (WakeupException e) { 
    // 忽略异常，正在关闭消费者
} catch (Exception e) {
     log.error("Unexpected error", e);
} finally {
         try {
            // 关闭时同步提交
            consumer.commitSync(currentOffsets); 
         } finally {
            consumer.close();
            System.out.println("Closed consumer and we are done"); 
         }
}

通过subscribe订阅的时候，传入ConsumerRebalanceListener的实现。在onPartitionsRevoked中同步提交偏移量，防止偏移量的丢失。

到指定的偏移位置开始消费

public class SaveOffsetsOnRebalance implements ConsumerRebalanceListener {
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        commitDBTransaction();
    }
    
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for(TopicPartition partition: partitions){
            consumer.seek(partition, getOffsetFromDB(partition));
        } 
    }
}

consumer.subscribe(topics, new SaveOffsetOnRebalance(consumer)); 
consumer.poll(0);

// 指定偏移位置
for (TopicPartition partition: consumer.assignment()) 
    consumer.seek(partition, getOffsetFromDB(partition));

while (true) {
     ConsumerRecords<String, String> records = consumer.poll(100);
     for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
        storeRecordInDB(record); 
        storeOffsetInDB(record.topic(), record.partition(), record.offset());
     }
     commitDBTransaction();
}

在partition重新分配时，读取保存的偏移位置，seek到指定的位置。

优雅地退出wakeup()

// 主线程处理时，需要在Runtime退出时调用wakeup()
Runtime.getRuntime().addShutdownHook(new Thread() { 
    public void run() {
        System.out.println("Starting exit..."); 
        consumer.wakeup();
        try {
            mainThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace(); 
        }
    } 
});

try {
    // 循环，直到按下Ctrl+C键，关闭的钩子会在退出时进行清理 
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        System.out.println(System.currentTimeMillis() + " -- waiting for data...");
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
        }

        for (TopicPartition tp: consumer.assignment())
            System.out.println("Committing offset at position:" + consumer.position(tp));
        consumer.commitSync(); 
    }
} catch (WakeupException e) { 
    // 忽略关闭异常
} finally {
    consumer.close();
    System.out.println("Closed consumer and we are done");
}

非订阅式消费

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

if (partitionInfos != null) {
    for (PartitionInfo partition : partitionInfos)
        partitions.add(new TopicPartition(partition.topic(), partition.partition()));
    
    consumer.assign(partitions); ➋
    while (true) {
         ConsumerRecords<String, String> records = consumer.poll(1000);
         for (ConsumerRecord<String, String> record: records) {             
              System.out.println("topic = %s, partition = %s, offset = %d,
customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
         }
         consumer.commitSync(); 
    }
}

通过partitionFor("topicName")获取指定主题的所有分区，通过Assign分配给自己所消费的分区。此模式和subacribe在同一消费者中，仅存在一个。