Kafka、Spring的Kafka与重传旧信息

Question

spring.kafka.bootstrap-servers=${kafka.host}:${kafka.port}
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.group-id=postfenix
spring.kafka.consumer.enable-auto-commit=false

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Post> postKafkaListenerContainerFactory(KafkaProperties kafkaProperties) {

    ConcurrentKafkaListenerContainerFactory<String, Post> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.getContainerProperties().setAckMode(AckMode.MANUAL);
    factory.setConsumerFactory(postConsumerFactory(kafkaProperties));

    return factory;
}

@KafkaListener(topics = "${kafka.topic.post.send}", containerFactory = "postKafkaListenerContainerFactory")
public void sendPost(ConsumerRecord<String, Post> consumerRecord, Acknowledgment ack) {

    Post post = consumerRecord.value();

    // do some logic here

    ack.acknowledge();
}

Answer 1

当使用Kafka时，客户端需要自己提交抵消。这与其他消息代理（如AMQP代理）形成鲜明对比，在其他消息代理中，代理跟踪客户机已经接收到的消息。

在您的情况下，您不会自动提交偏移量，因此Kafka希望您手动提交这些偏移量（因为此设置:spring.Kafka.consumer.enable-auto-commit=false)。如果不在程序中手动提交偏移量，则所描述的行为与预期的基本一致。卡夫卡根本不知道你的程序成功地处理了什么消息。每次重新启动程序时，Kafka都将看到程序尚未提交任何偏移，并将应用spring.Kafka.consumer.auto-offset-reset=aresty中提供的策略，这意味着队列中的第一条消息。

如果这对您来说都是新的，我建议阅读关于Kafka的这篇文档和这篇Spring文档，因为Kafka与其他消息代理有很大的不同。