Kafka Consumer 简易配置

缺乏、安全感 2023-01-17 11:55 113阅读 0赞

@Bean(value = "kafkaConsumer", destroyMethod = "close")
    public KafkaConsumer<String, String> getKafkaConsumer() throws Exception {
        Properties properties = new Properties();
      properties.putAll(xxx.getClientConfigs(kafkaClientId,kafkaUsername,kafkaPassword));
        //不设置默认值是latest（第一次消费或者越界从最新开始消费）；earliest：第一次消费或者越界从最小位点开始消费数据
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        //是否自动提交位点,默认是true.默认的自动提交位点的时间间隔是5000ms，false的情况下是需要用户自己调用commit方法自己手动提交位点信息的
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties, new StringDeserializer(), new StringDeserializer());
        kafkaConsumer.subscribe(Collections.singleton(kafkaTopic));
        return kafkaConsumer;
    }

## Kafka Consumer ##

kafka消费的配置如下:

*  bootstrap.servers： kafka的地址。
 *  group.id：组名 不同组名可以重复消费。例如你先使用了组名A消费了kafka的1000条数据，但是你还想再次进行消费这1000条数据，并且不想重新去产生，那么这里你只需要更改组名就可以重复消费了。
 *  enable.auto.commit：是否自动提交，默认为true。
 *  auto.commit.interval.ms: 从poll(拉)的回话处理时长。
 *  session.timeout.ms:超时时间。
 *  max.poll.records:一次最大拉取的条数。
 *  auto.offset.reset：消费规则，默认earliest 。 earliest: 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 。 latest: 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 。 none: topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常。
 *  key.serializer: 键序列化，默认org.apache.kafka.common.serialization.StringDeserializer。
 *  value.deserializer:值序列化，默认org.apache.kafka.common.serialization.StringDeserializer。

## 消息提交主要有以下两种方式 ##

**自动提交：enable.auto.commit 可以配置自动提交时间**

可配合 auto.commit.interval.ms 设置自动提交频率

可通过 enable.auto.commit=false关闭自动提交

重复消费问题：设置自动提交后，若在提交间隔Consumer挂了，就可能出现已经被消费了的消息由于未被确认而再次被消费；

**手动提交：即使存在重复提交，也可以有有效控制重复消息数**

异步提交：consumer.commitAsync() 同步提交：consumer.commitSync()，存在阻塞，因此一般批量提交，以减少阻塞频率

**enable.auto.commit：false**

自动提交offset下标。 但是offset下标自动提交其实在很多场景都不适用，因为自动提交是在kafka拉取到数据之后就直接提交，这样很容易丢失数据，尤其是在需要事物控制的时候。 很多情况下我们需要从kafka成功拉取数据之后，对数据进行相应的处理之后再进行提交。如拉取数据之后进行写入mysql这种 ， 所以这时我们就需要进行手动提交kafka的offset下标。

**offset**：指的是kafka的topic中的每个消费组消费的下标。 简单的来说就是一条消息对应一个offset下标，每次消费数据的时候如果提交offset，那么下次消费就会从提交的offset加一那里开始消费。 比如一个topic中有100条数据，我消费了50条并且提交了，那么此时的kafka服务端记录提交的offset就是49(offset从0开始)，那么下次消费的时候offset就从50开始消费。

kafka的offset下标的记录实际会有两份，服务端会自己记录一份，本地的消费者客户端也会记录一份，提交的offset会告诉服务端已经消费到这了，但是本地的并不会因此而改变offset进行再次消费。

简单的来说假如有10条数据，在第5条的时候进行提交了offset下标，那么服务端就知道该组消费的下标到第5条了，如果同组其他的consumer进行消费的时候就会从第6条开始进行消费。但是本地的消费者客户端并不会因此而改变，它还是会继续消费下去，并不会再次从第6条开始消费，所以会出现上图情况。

參考：[https://www.cnblogs.com/xuwujing/p/8432984.html][https_www.cnblogs.com_xuwujing_p_8432984.html]

[https_www.cnblogs.com_xuwujing_p_8432984.html]: https://www.cnblogs.com/xuwujing/p/8432984.html