Kafka2.0消费者客户端使用

2023年7月5日 129次阅读

1 初始化配置

  Kafka 通过 KafkaConsumer 构造器初始化生产者客户端的配置。
  常用的重要配置，详见官网。

  • bootstrap.servers：Kafka 集群地址（host1:post,host2:post），Kafka 客户端初始化时会自动发现地址，所以可以不填写所有地址。
  • group.id：消费组 ID
  • key.serializer：实现了 Kafka 序列化接口的类，用来序列化 key。
  • value.serializer：实现了 Kafka 序列化接口的类，用来序列化 value。
  • enable.auto.commit：默认 true，表示消费者偏移量会定期提交到后台。
  • auto.offset.reset：Kafka 的偏移量。
     earliest：自动重置为最早的偏移量。
     latest：自动重置为最新的偏移量。
     none：如果没有找到消费组之前的那个偏移量，则消费者抛出异常。
     其他：消费者抛出异常。
  • fetch.min.bytes/fetch.max.bytes：消费者一次拉取的最小/最大值。
  • max.poll.interval.ms：消费者拉取的最大间隔时间，超时后从组中移除消费者。
  • heartbeat.interval.ms：心跳发送间隔的超时时间，超时后从组中移除消费者。
  • isolation.level：事务的隔离级别。
     read_uncommitted：默认，可以消费到所有消息，包括被中止的消息。
     read_committed：只能消费到事务提交过的消息。
     非事务性消息无条件返回。
// 基础配置
Map<String, Object> configs = new HashMap<>();
configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
configs.put(ConsumerConfig.GROUP_ID_CONFIG, "my_test");
configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configs);

2 订阅主题

  Kafka 消费者提供4种方式订阅主题，1种方式指定分区。

  • topics：指定主题集。
  • pattern：指定正则表达式来匹配主题。
  • listener：消费者再均衡监听器。
  • partitions：指定分区集合。
// 指定主题
public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener)
public void subscribe(Collection<String> topics)
public void subscribe(Pattern pattern, ConsumerRebalanceListener listener)
public void subscribe(Pattern pattern)
// 指定分区
public void assign(Collection<TopicPartition> partitions)

3 拉取消息

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configs);
consumer.subscribe(Collections.singletonList("test")); // 指定主题
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));

 3.1 指定偏移量消费

TopicPartition tp = new TopicPartition("test", 0);
consumer.assign(Collections.singletonList(tp)); // 订阅指定分区
consumer.seek(tp, 4L); // 指定分区偏移量值为4
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));

 3.2 指定时间戳消费

        TopicPartition tp = new TopicPartition("test", 0);
        consumer.assign(Collections.singletonList(tp)); // 订阅指定分区
        Map<TopicPartition, Long> tpTime = new HashMap<>();
        tpTime.put(tp, 1563027475113L); // 指定时间戳
        Map<TopicPartition, OffsetAndTimestamp> tpOffsetAndTime = consumer.offsetsForTimes(tpTime);
        long offset = tpOffsetAndTime.get(tp).offset(); // 获取偏移量
        consumer.seek(tp, offset); // 指定偏移量
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));

4 提交偏移量

 4.1 同步提交

  参数说明

  • offsets：可以指定提交分区的偏移量。
  • timeout：偏移量提交成功的超时时间。
public void commitSync()
public void commitSync(Duration timeout)
public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets)
public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets, final Duration timeout)

 4.2 异步提交

  参数说明

  • offsets：可以指定提交分区的偏移量。
  • callback：异步回调。
public void commitAsync()
public void commitAsync(OffsetCommitCallback callback)
public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)

5 其他方法

// 获取分配给当前消费者的分区集合
public Set<TopicPartition> assignment()
// 取消订阅
public void unsubscribe()
// 找到指定分区的第一个偏移量
public void seekToBeginning(Collection<TopicPartition> partitions)
// 找到指定分区的最后一个偏移量
public void seekToEnd(Collection<TopicPartition> partitions)
// 获取指定分区即将消费的下一个偏移量
public long position(TopicPartition partition)
// 获取指定分区最后提交的偏移量
public OffsetAndMetadata committed(TopicPartition partition)
// 获取指定主题的分区列表
public List<PartitionInfo> partitionsFor(String topic)
// 获取所有主题的信息
public Map<String, List<PartitionInfo>> listTopics()
// 暂停消费
public void pause(Collection<TopicPartition> partitions)
// 恢复被暂停的消费
public void resume(Collection<TopicPartition> partitions)
// 获取暂停的分区列表
public Set<TopicPartition> paused()
// 获取指定分区第一个偏移量
public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions)
// 获取指定分区最后一个偏移量
public Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions)
// 唤醒消费者
public void wakeup()
点赞