Kafka消费者不会重试处理失败的消息

首先，需要确保Kafka消费者设置了重试策略。可以通过设置属性max.poll.retries和重试间隔时间的属性retry.backoff.ms来实现。例如：

// 创建Kafka消费者配置 Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test"); props.put("auto.offset.reset", "earliest"); props.put("enable.auto.commit", "false"); props.put("max.poll.records", "100"); props.put("max.poll.retries", "3"); // 设置重试次数 props.put("retry.backoff.ms", "5000"); // 设置重试间隔时间

// 创建Kafka消费者 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props, new StringDeserializer(), new StringDeserializer()); consumer.subscribe(Collections.singletonList("my-topic"));

// 消费消息 while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { try { // 处理消息 processRecord(record); // 手动提交偏移量 consumer.commitSync(); } catch (Exception e) { // 异常处理 e.printStackTrace(); } } }

其次，在处理消息时，需要捕获异常并将消费者的当前偏移量重置为上一次成功消费的偏移量。这样，消费者会重新拉取并重试处理失败的消息。例如：

private void processRecord(ConsumerRecord<String, String> record) { try { // 处理消息 System.out.printf("topic = %s, partition = %d, offset = %d, key = %s, value = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value()); } catch (Exception e) { // 重置偏移量 long lastOffset = record.offset() - 1; TopicPartition partition = new TopicPartition(record.topic(), record.partition()); Map<TopicPartition, OffsetAndMetadata> offsetMap = Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset)); consumer.commitSync(offsetMap); } }