kafka的Java客户端消费者

kafka的Java客户端消费者相关配置

一.消费者的基本实现

public class MyConsumer {
    private final static String TOPIC_NAME = "my-replicated-topic";
    private final static String CONSUMER_GROUP_NAME = "testGroup";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
// 消费分组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
//1.创建⼀个消费者的客户端
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
//2.消费者订阅主题列表
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        while (true) {
            /*
             * 3.poll() API 是拉取消息的⻓轮询 
             */
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                //4.打印消息
                System.out.printf("收到消息：partition = %d,offset = %d, key =
                        % s, value = % s % n ", record.partition(),
                record.offset(), record.key(), record.value());
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

二.关于消费者自动提交和手动提交的offset

1.提交的内容

• 消费者无论是自动提交还是手动提交,都需要把所属的小肥猪+消费的某个主题+消费的某个分区及消费的偏移量,这些信息提交到集群的_consumer_offset主题里面

2.自动提交

• 消费者将消息poll下来以后就会自动提交offset
• 自动提交相关参数配置

//是否自动提交offset,默认是true
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
//自动提交offset的时间
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
1
2
3
4

• 注意:自动提交会丢失消息,因为消费者在消费前提交offset,有可能提交完毕之后还没有消费时消费者就挂掉了.

3.手动提交

• 修改配置屏蔽自动提交

//是否自动提交offset,设置为false
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
1
2

3.1手动同步提交

• 在消费完后调用同步提交方法,当集群返回ack前一直阻塞,返回ack后表示成功,执行之后的逻辑

if (records.count() > 0) {
// ⼿动同步提交offset，当前线程会阻塞直到offset提交成功 // ⼀般使⽤同步提交，因为提交之后⼀般也没有什么逻辑代码了 consumer.commitSync();
}
1
2
3

3.2手动异步提交

• 在消息消费完毕之后进行提交,不需要等待集群ack,直接执行之后的逻辑,可以设置一个回调方法,提供集群调用

        if (records.count() > 0) {
// ⼿动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后⾯的程序 逻辑
            consumer.commitAsync(new OffsetCommitCallback() {
                @Override
                public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                    if (exception != null) {
                        System.err.println("Commit failed for " + offsets);
                        System.err.println("Commit failed exception: " +
                                exception.getStackTrace());
                    }
                }
            });
        }
1
2
3
4
5
6
7
8
9
10
11
12
13

三.消费者poll消息的细节

1.长轮询poll消息

• 默认情况下,消费者一次会poll500条消息.

//一次poll最大拉取消息的条数,可以根据消费速度的快慢来进行设置
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
1
2

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

        while (true) {
            /*
             * 3.poll() API 是拉取消息的⻓轮询 
             */
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                //4.打印消息
                System.out.printf("收到消息：partition = %d,offset = %d, key =
                        % s, value = % s % n ", record.partition(),
                record.offset(), record.key(), record.value());
            }
1
2
3
4
5
6
7
8
9
10
11

意味着:

• 如果一次poll到500条,就直接执行for循环
• 如果一次没有poll到500条,且时间在1s之内,那么长轮询继续poll要么到500条,要么到1s
• 如果多次poll都没有达到500条,且1s时间到了直接执行for循环

如果两次poll的时间间隔超过了30s,集群会认为消费者的消费能力过弱,该消费者被踢出消费组,触发rebalance机制,rebalance机制会造成性能开销.可以通过这个参数,让一次poll的消息条数过少一些.

//让一次poll最大拉取消息的条数,可以根据消费速度快慢来进行设置
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
//如果两次poll的时间如果超出了30s间隔时间,kafka会认为其消费能力过弱,将其踢出消费组,将分区分配给其他消费者. -rebalance
props.put(ConsumerConfig.MAX_POLL_TNTERVAL_MS_CONFIG, 30 * 1000);
1
2
3
4

2.消费者健康状况的检查

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

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

3.指定分区,偏移量,时间消费

• 指定分区消费

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

• 从头消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
1
2

• 指定offset消费

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

• 指定时间消费

根据时间取所有的partition中去确定该事件对应的offset,然后去所有的partition中找到该offset之后的消息进行消费.

        List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);
//从1⼩时前开始消费
        long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;
        Map<TopicPartition, Long> map = new HashMap<>();
        for (PartitionInfo par : topicPartitions) {
            map.put(new TopicPartition(TOPIC_NAME, par.partition()),
                    fetchDataTime);
        }
        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);
            System.out.println();
//根据消费⾥的timestamp确定offset if (value != null) {
            consumer.assign(Arrays.asList(key));
            consumer.seek(key, offset);
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

四.新消费组的消费offset规则

• 新消费组中的消费者在启动之后,默认会从当前分区的最后一条消息的offset + 1开始消费(消费新消息).可以通过以下设置,让新的消费者第一次从头开始消费.之后开始消费新消息(最后消费的位置的偏移量 + 1)

  • Latest:默认的,消费新消息
  • earlist:第一次从都开始进行消费,之后开始消费新消息(最后消费的位置的偏移量 + 1)

  props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  1