Kafka消息重试与错误处理策略-CFANZ编程社区

背景

Kafka是一个高性能、高可靠的分布式消息系统，被广泛应用于大规模数据处理、日志收集等场景。在实际应用中，由于各种原因，消息可能会发送失败或者处理失败，这时候就需要考虑消息重试和错误处理策略。

消息重试

消息重试是指在消息发送或者处理失败后，重新发送消息的过程。Kafka提供了两种消息重试的方式：

1. 自动重试

Kafka的生产者API提供了自动重试的功能，当消息发送失败时，生产者会自动重试，直到消息发送成功或者达到最大重试次数。默认情况下，最大重试次数为3次。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 3);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String, String> producer = new KafkaProducer<>(props);

for (int i = 0; i < 100; i++) {
    producer.send(new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i)));
}

producer.close();

2. 手动重试

手动重试是指在消息发送或者处理失败后，由应用程序自行处理重试逻辑。在实际应用中，手动重试通常会结合定时任务或者消息队列等机制，实现更加灵活的重试策略。

public void send(String topic, String message) {
    ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
    try {
        producer.send(record).get();
    } catch (InterruptedException | ExecutionException e) {
        // 发送失败，进行重试
        retry(record);
    }
}

private void retry(ProducerRecord<String, String> record) {
    // TODO: 实现重试逻辑
}

错误处理

错误处理是指在消息发送或者处理失败后，如何处理错误。Kafka提供了两种错误处理的方式：

1. 抛出异常

Kafka的生产者API和消费者API都会抛出异常，应用程序可以捕获这些异常并进行相应的处理。

public void send(String topic, String message) throws Exception {
    ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
    try {
        producer.send(record).get();
    } catch (InterruptedException | ExecutionException e) {
        // 发送失败，抛出异常
        throw new Exception("Failed to send message", e);
    }
}

public void consume() throws Exception {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            try {
                // 处理消息
            } catch (Exception e) {
                // 处理失败，抛出异常
                throw new Exception("Failed to process message", e);
            }
        }
    }
}

2. 记录日志

除了抛出异常外，应用程序还可以记录日志，以便后续排查问题。

public void send(String topic, String message) {
    ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
    try {
        producer.send(record).get();
    } catch (InterruptedException | ExecutionException e) {
        // 发送失败，记录日志
        logger.error("Failed to send message", e);
    }
}

public void consume() {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            try {
                // 处理消息
            } catch (Exception e) {
                // 处理失败，记录日志
                logger.error("Failed to process message", e);
            }
        }
    }
}