Kafka学习笔记七：消费者配置详解

1.长轮询poll消息

默认情况下，消费者一次会poll条消息

props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);

代码中设置了长轮询的时间是1000毫秒

while (true){
    /**
     * poll是拉取消息的长轮询
     */
    ConsumerRecords<String,String> records =  consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("收到消息：partition = %d,offset = %d,key = %s ,value = %s%n",
                record.partition(),record.offset(),record.key(),record.value());
    }
 }

• 如果一次poll到500条消息，就直接执行for循环
• 如果一次没有poll到500条，且时间在1秒内，那么长轮询继续poll，要么到500条，要么到1s
• 如果两次poll的时间超过了30s的时间间隔，kafka会认为消费者消费能力过弱，将其踢出消费者组，将分区分配给其他消费者rebalance

props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 30 * 1000);

2.消费者健康状态检查：

消费者每隔1s向kafka集群发送心跳，集群发现如果有超过10s没有续约的消费者，将被踢出消费组，出发消费组的rebalance机制，将该分区交给消费组里其他消费者进行消费

//consumer给broker发送心跳的间隔时间
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
//kafka如果超过10秒没有收到消费这的心跳，会把消费者踢出消费者组，进行rebalance,把分区分配给其他消费者
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);

3.指定分区消费：

//指定分区消费
consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));

4.消息回溯消费：

从topic offset的0号位置开始消费

//消息回溯消费
consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));
consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));

5.指定offset消费：

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));
consumer.seek(new TopicPartition(TOPIC_NAME,0),10);

6.从指定时间点开始消费：

//从指定时间点开始消费
//1. 拿到主题下所有的分区
List<PartitionInfo> partitionInfos = consumer.partitionsFor(TOPIC_NAME);
//2. 拿到1小时前的时间
long fetchDateTime = new Date().getTime()- 1000 * 60 * 60;
HashMap<TopicPartition, Long> map = new HashMap<>();
for (PartitionInfo partitionInfo : partitionInfos) {
    map.put(new TopicPartition(TOPIC_NAME, partitionInfo.partition()),fetchDateTime);
}
//3. consumer.offsetsForTimes 根据时间拿到偏移量
Map<TopicPartition, OffsetAndTimestamp> parmap = consumer.offsetsForTimes(map);
for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parmap.entrySet()) {
    //先拿到分区
    TopicPartition key = entry.getKey();
    OffsetAndTimestamp value = entry.getValue();
    if(key ==null || value == null) continue;
    long offset = value.offset();
    System.out.println("partition-"+key.partition() + "|offset-"+offset);
    //根据消费里的timestamp确定offset
    if(value != null){
        consumer.assign(Arrays.asList(key));
        consumer.seek(key,offset);
    }
}

7.新消费组的消费偏移量：

/**
 * 当消费主题的是一个新的消费组，或者指定offset的消费方式，offset不存在
 *  latest(默认)：只消费自己启动后发送到主题的消息
 *  earliest: 第一次从头开始消费，以后按照消费offset记录继续消费，这个需要区别于consumer.seekToBeginning(每次从头开始消费)
 */
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");