Kafka的不同消费方式和消费者代码示例

快来打我* 2023-02-10 05:17 12阅读 0赞

# 一、前言 #

本文针对 Kafka 的消费者代码，均由 Kafka 原生的客户端 API 构成，而非 spring-kafka 或 SpringCloudStream 等对于原生 API 进行封装的示例。

Kafka 的消费方式，根据不同的需求可以做出多样的选择。

*  可以根据是否手动指定分区，而分为 **Subscribe** 和 **Assign** 这两种模式
 *  可以根据消费特定条件，而分为**最新偏移量**、**指定偏移量**和**指定时间戳**开始消费这三种方式

# 二、各种消费方式 #

## 2.1 Subscribe模式（订阅主题） ##

对 Kafka 消费者来说，***Subscribe*** 模式是最简单的方式，往往也是最常用的方式，即仅需要订阅一个主题即可。

public static void testSubscribe() {
        Properties properties = new Properties();
        // Kafka集群地址
        properties.put("bootstrap.servers", "100.1.4.16:9092,100.1.4.17:9092,100.1.4.18:9092");
        // 消费者组，仅在subscribe模式下生效，用于分区自动再均衡，而assign模式直接指定分区
        properties.put("group.id", "test_group");
        // 反序列化器
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        // 订阅topic
        String topic = "test_topic";
        consumer.subscribe(Pattern.compile(topic));
        while (true) {
            // 每1000ms轮询一次
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));
            log.info("本次轮询到：{}条", records.count());
            for (ConsumerRecord<String, String> record : records) {
                log.info("-------消息来了：topic={}, partition={}, offset={}, value={}", record.topic(), record.partition(),
                        record.offset(), record.value());
            }
        }
    }

### 2.1.1 解析 ###

通过 Kafka 原生的消费者 API 来消费数据，主要分为三个步骤：

1.  配置必要信息以构造消费者实例
2.  订阅主题（或后文中的指定分区）
3.  轮询消息

### 2.1.2 注意事项 ###

1.  一般情况下，配置 *group.id* 仅在 ***Subscribe*** 模式下生效，一般认为消费者组的概念主要对消息分区自动再均衡起作用。
2.  一个消费者可以消费多个主题和多个分区，但一个分区只能同时被同一个消费者组里的一个消费者消费。
3.  通过 *public void subscribe(Pattern pattern, ConsumerRebalanceListener listener)*，可监听消费者组内的分区再均衡，进而实现自定义的业务。

## 2.2 Assign模式（手动指定分区） ##

上面说了，除了 ***Subscribe*** 模式，还有 ***Assign*** 模式，用来手动指定要消费的消息分区。

public static void testAssignOffset() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "100.1.4.16:9092,100.1.4.17:9092,100.1.4.18:9092");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // 默认每次轮询最多取多少条消息，默认500
        properties.put("max.poll.records", 1);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        String topic = "test_topic";
        TopicPartition tp = new TopicPartition(topic, 0);
        // 指定分区
        consumer.assign(Collections.singletonList(tp));
        log.info("本topic下所有的分区：{}", consumer.partitionsFor(topic));
        // 获取消费者被分配到的分区（注意，assign模式会直接返回手动指定的分区，而subscribe模式等到自动分配分区后才有返回）
        log.info("本消费者分配到的分区：{}", consumer.assignment());
        // 为某个指定分区任意位置、起始位置、末尾位置为起始消费位置(offset默认从0开始)
        // 注意若分配的offset<分区最小的offset(可能kafka删除策略影响，比如默认删除超过7d的数据导致最小offset值变化)，将从最新offset处监听消费
        // consumer.seek(tp, 5);
        // consumer.seekToBeginning(Arrays.asList(tp));
        // consumer.seekToEnd(Collections.singletonList(tp));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));
            log.info("本次轮询到：{}条", records.count());
            for (ConsumerRecord<String, String> record : records) {
                log.info("-------消息来了：topic={}, partition={}, offset={}, value={}", record.topic(), record.partition(),
                        record.offset(), record.value());
            }
        }
    }

### 2.2.1 解析 ###

*  在 ***Assign*** 模式下，消费者不再需要消费者组的概念，所以 *group.id* 可以忽略配置。
 *  通过消费者配置 *max.poll.records*，可修改每次轮询的最大消息数，默认值 500。
 *  与 ***Subscribe*** 模式不同，***Assign*** 模式通过 *public void assign(Collection<TopicPartition> partitions)*，来指定数据分区集合。
 *  通过 *public List<PartitionInfo> partitionFor(String topic)*，可获取某主题下的所有分区信息。
 *  通过 *public Set<TopicPartition> assignment()*，可获取消费者被分配到的分区集合。

### 2.2.2 注意事项 ###

***Assign*** 模式下，可通过 *seek/seekToBeginning/seekToEnd* 等 API 来指定偏移量 offset 开始消费。

## 2.3 指定偏移量消费 ##

在 2.2 上一节的代码基础上，打开 *seek/seekToBeginning/seekToEnd* 等注释，即可指定偏移量进行消费。

### 2.3.1 注意事项 ###

1.  若指定分区的偏移量已在分区上不存在（比如受到 Kafka 清除策略的影响），则将从最新 offset 处监听消费。
2.  因为不是 ***Subscribe*** 模式，所以不存在消费者组的概念，所以即便设置了消费者组，也不会触发消费者组的分区再均衡操作。

## 2.4 指定时间戳消费 ##

Kafka 不仅支持指定偏移量消费，也支持指定消息的时间戳进行消费。不过根本上也是通过偏移量的消费。

public static void testAssignTimeStamp() throws ParseException {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "100.1.4.16:9092,100.1.4.17:9092,100.1.4.18:9092");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        String topic = "test_topic";
        // 设置消费起始时间
        String startTime = "2020-05-19 15:52:41";
        Long startTimestamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(startTime).getTime();
        Map<TopicPartition, Long> timestampMap = new HashMap<>();
        // 获取每一个分区信息
        List<PartitionInfo> partitionInfoLst = consumer.partitionsFor(topic);
        for (PartitionInfo partitionInfo : partitionInfoLst) {
            // 设置每一个分区的起始消费时间为指定时间
            timestampMap.put(new TopicPartition(partitionInfo.topic(), partitionInfo.partition()), startTimestamp);
        }
        // 通过时间戳查找给定分区的偏移量
        Map<TopicPartition, OffsetAndTimestamp> offsetMap = consumer.offsetsForTimes(timestampMap);
        // 指定分区
        consumer.assign(offsetMap.keySet());
        // 设置每一个分区的指定时间对应的消费偏移量
        for (TopicPartition topicPartition : offsetMap.keySet()) {
            consumer.seek(topicPartition, offsetMap.get(topicPartition).offset());
        }
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));
            log.info("本次轮询到：{}条", records.count());
            for (ConsumerRecord<String, String> record : records) {
                log.info("-------消息来了：topic={}, partition={}, offset={}, value={}", record.topic(), record.partition(),
                        record.offset(), record.value());
            }
        }
    }

### 2.4.1 解析 ###

指定时间戳消费的关键步骤如下：

1.  配置必要信息以构造消费者实例
2.  对要消费的各个主题下的各个分区，设置开始消费的时间戳
3.  根据各自分区的时间戳通过 *offsetsForTimes* 对应获取各自的偏移量
4.  指定消费的消息分区集合
5.  指定各自分区开始消费的偏移量
6.  轮询消息

### 2.4.2 注意事项 ###

1.  若指定分区的时间戳对应的偏移量已在分区上不存在（比如受到 Kafka 清除策略的影响），则将从最新 offset 处监听消费。
2.  因为不是 ***Subscribe*** 模式，所以不存在消费者组的概念，所以即便设置了消费者组，也不会触发消费者组的分区再均衡操作。

## 2.5 Subscribe模式下指定偏移量消费 ##

上面的指定偏移量也好，指定时间戳的消费方式也罢，都是属于 ***Assign*** 模式的。那 ***Subscribe*** 模式能否也可以指定偏移量消费呢？答案是可以的。

public static void testSubscribeOffset() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "100.1.4.16:9092,100.1.4.17:9092,100.1.4.18:9092");
        properties.put("group.id", "test_group");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        String topic = "test_topic";
        Map<TopicPartition, OffsetAndMetadata> hashMaps = new HashMap<TopicPartition, OffsetAndMetadata>();
        hashMaps.put(new TopicPartition(topic, 0), new OffsetAndMetadata(5));
        // 手动提交指定offset作为起始消费offset
        consumer.commitSync(hashMaps);
        consumer.subscribe(Pattern.compile(topic));
        while (true) {
            // 每1000ms轮询一次
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));
            log.info("本次轮询到：{}条", records.count());
            for (ConsumerRecord<String, String> record : records) {
                log.info("-------消息来了：topic={}, partition={}, offset={}, value={}", record.topic(), record.partition(),
                        record.offset(), record.value());
            }
        }
    }

### 2.5.1 解析 ###

实际上，我们是通过在开启轮询之前，手动提交一次偏移量信息，然后再去轮询消息的方式达到目的。

### 2.5.2 注意事项 ###

同样地，若指定分区的偏移量已在分区上不存在（比如受到 Kafka 清除策略的影响），则将从最新 offset 处监听消费。