kafka-消费者消费消息的基本实现

目录

消费者代码基本实现消费者自动提交和手动提交offset消息者poll消息细节消费者的健康状态检查消息者指定分区和偏移量、时间消费消息者新消费组的消费offset规则

消费者代码基本实现

public class myConsumer { private final static String TOPIC_NAME = "myTopic"; private final static String CONSUMER_GROUP_NAME = "testGroup"; public static void main(String[] args) { Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.72.128:9092,192.168.72.128:9093,192.168.72.128: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()); //关闭自动提交offset props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false); //创建⼀个消费者的客户端 KafkaConsumer consumer = new KafkaConsumer(props); // 消费者订阅主题列表 consumer.subscribe(Arrays.asList(TOPIC_NAME)); while (true) { ConsumerRecords records = consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecord record : records) { System.out.printf("收到消息：partition = %d,offset = %d, key =%s, value = %s%n", record.partition(), record.offset(), record.key(), record.value()); } } }}

消费者自动提交和手动提交offset

1、提交的内容

消费者⽆论是⾃动提交还是⼿动提交，都需要把所属的消费组+消费的某个主题+消费的某个分区以及消费的偏移量，这样的信息提交到集群的_consumer_offsets主题里面。

2、自动提交
消费者poll消息下来以后就会⾃动提交offset。

// 是否⾃动提交offset，默认就是trueprops.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");// ⾃动提交offset的间隔时间props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

注意：⾃动提交会丢消息。因为消费者在消费前提交offset，有可能提交完后还没消费时消费
者挂了。

3、手动提交
把⾃动提交的配置改成false即可。

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

需要注意的是：手动提交 offset 的方法有两种：分别是 commitSync（同步提交）和 commitAsync（异步提交）。两者的相同点是，都会将本次 poll 的一批数据最高的偏移量提交；不同点是，commitSync 阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而 commitAsync 则没有失败重试机制，故有可能提交失败。

①手动同步提交
在消费完消息后调⽤同步提交的⽅法，当集群返回ack前⼀直阻塞，返回ack后表示提交
成功，执⾏之后的逻辑。代码如下：

while (true) { ConsumerRecords records = consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecord record : records) { System.out.printf("收到消息：partition = %d,offset = %d, key= %s, value = %s%n", record.partition(), record.offset(), record.key(), record.value()); } //所有的消息已消费完 if (records.count() > 0) {//有消息 // ⼿动同步提交offset，当前线程会阻塞直到offset提交成功 // ⼀般使⽤同步提交，因为提交之后⼀般也没有什么逻辑代码了 consumer.commitSync();//=======阻塞=== 提交成功 }}

②手动异步提交
在消息消费完后提交，不需要等到集群ack，直接执⾏之后的逻辑，可以设置⼀个回调⽅
法，供集群调⽤。

while(true){ ConsumerRecords records=consumer.poll(Duration.ofMillis(1000)); for(ConsumerRecord record:records) { System.out.printf("收到消息：partition = %d,offset = %d, key =%s,value=%s%n", record.partition(), record.offset(),record.key(),record.value()); } //所有的消息已消费完 if(records.count()>0){ // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面的程序逻辑 consumer.commitAsync(new OffsetCommitCallback() { @Override public void onComplete(Map offsets,Exception exception) { if(exception!=null){ System.err.println("Commit failed for "+offsets); System.err.println("Commit failed exception: "+ exception.getStackTrace()); } } }); }}

消息者poll消息细节

1、默认情况下，消费者⼀次会poll500条消息。

//⼀次poll最⼤拉取消息的条数，可以根据消费速度的快慢来设置props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);

2、代码中设置了⻓轮询的时间是1000毫秒

ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));

意味着：

如果⼀次poll到500条，就直接执⾏for循环。如果这⼀次没有poll到500条。且时间在1秒内，那么⻓轮询继续poll，要么到500条，要么到1s。如果多次poll都没达到500条，且1秒时间到了，那么直接执⾏for循环。如果两次poll的间隔超过30s，集群会认为该消费者的消费能⼒过弱，该消费者被踢出消费组，触发rebalance机制，rebalance机制会造成性能开销。可以通过设置这个参数，让⼀次poll的消息条数少⼀点

//⼀次poll最⼤拉取消息的条数，可以根据消费速度的快慢来设置 props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500); //如果两次poll的时间如果超出了30s的时间间隔，kafka会认为其消费能⼒过弱，将其踢 //出消费组。将分区分配给其他消费者。-rebalance props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);

消费者的健康状态检查

消费者每隔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)

消息者指定分区和偏移量、时间消费

指定分区消费

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

从头消费

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

指定offset消费

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

指定时间消费

根据时间，去所有的partition中确定该时间对应的offset，然后去所有的partition中找到该offset之后的消息开始消费。

List topicPartitions = consumer.partitionsFor(TOPIC_NAME); //从1小时前开始消费 long fetchDataTime = new Date().getTime() - 1000 * 60 * 60; Map map = new HashMap<>(); for (PartitionInfo par : topicPartitions) { map.put(new TopicPartition(TOPIC_NAME, par.partition()),fetchDataTime); } Map parMap = consumer.offsetsForTimes(map); for (Map.Entry 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); } }

消息者新消费组的消费offset规则

新消费组中的消费者在启动以后，默认会从当前分区的最后⼀条消息的offset+1开始消费（消费新消息）。可以通过以下的设置，让新的消费者第⼀次从头开始消费。之后开始消费新消息（最后消费的位置的偏移量+1）

Latest:默认的，消费新消息。

earliest：第⼀次从头开始消费。之后开始消费新消息（最后消费的位置的偏移量+1）。

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