在Kafka中获取最后一个事件-JAVA

Question

实际上，我用@KafkaListener来阅读一个主题中的事件，我想阅读100个事件，然后放一个线程。在一定时间内睡觉。我的问题是，当线程返回时，侦听器继续执行我读到的最后一个事件，但我想在线程睡眠时放弃这些事件，继续执行主题中的最后一个事件。

比如：

1-100-捕获

睡线

101-500

线程返回

501-601-捕获

101-500事件可以被丢弃

代码：

@KafkaListener(topics = "topic")
public void consumeBalance(ConsumerRecord<String, String> payload) throws InterruptedException {

    this.valorMaximoDeRequest = this.valorMaximoDeRequest + 1;
    if (this.valorMaximoDeRequest <= 100) {
        log.info("Encontrou evento )");
        log.info("Key: " + payload.key() + ", Value:" + payload.value());
        log.info("Partition:" + payload.partition() + ",Offset:" + payload.offset());

        JsonObject jsonObject = new Gson().fromJson(payload.value(), JsonObject.class);
        String accountId = jsonObject.get("accountId").getAsString();
        log.info(">>>>>>>>> accountId: " + accountId);

    } else {
        this.valorMaximoDeRequest = 0;
        Thread.sleep(60*1000);
    }

}

kafka配置：

@Bean
    public Map<String, Object> kafkaFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put("specific.avro.reader", Boolean.TRUE);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "brokers");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "1");

        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.put("security.protocol", "SASL_PLAINTEXT");
        return props;
    }

Answer 1

首先，您不应该强制监听线程Hibernate。消费者可能被认为已经死亡，并触发消费者重新平衡。您最好对消费者使用暂停和恢复。请参见https://docs.spring.io/spring-kafka/docs/current/reference/html/#pause-resume

然后，如果你想跳过当消费者睡着时发布的记录，你必须在消费者醒来时寻找。请参见https://docs.spring.io/spring-kafka/docs/current/reference/html/#seek但是这并不简单：当消费者活跃或不拥有分区时，Kafka消费者不允许你寻找。