Kafka消费者无限重试 | 扩展时的重新平衡问题

在Kafka消费者无限重试和扩展时的重新平衡问题中，可以使用以下代码示例来解决问题：

首先，创建一个带有自定义重试逻辑的消费者：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

public class RetryConsumer {

    private static final int MAX_RETRY_COUNT = 3;

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "retry-group");
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-topic"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                // 在重新平衡之前提交偏移量
                consumer.commitSync();
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                // 重新平衡之后从最新的偏移量开始消费
                consumer.seekToEnd(partitions);
            }
        });

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    // 处理消息
                    boolean success = processMessage(record);
                    if (success) {
                        // 消息处理成功，提交偏移量
                        consumer.commitAsync();
                    } else {
                        // 消息处理失败，进行重试逻辑
                        retryLogic(record, consumer);
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }

    private static boolean processMessage(ConsumerRecord<String, String> record) {
        // 处理消息的逻辑
        return true; // 返回true表示处理成功，返回false表示处理失败
    }

    private static void retryLogic(ConsumerRecord<String, String> record, KafkaConsumer<String, String> consumer) {
        int retryCount = 0;
        while (retryCount < MAX_RETRY_COUNT) {
            // 进行重试逻辑
            boolean success = processMessage(record);
            if (success) {
                // 重试成功，提交偏移量
                consumer.commitAsync();
                break;
            } else {
                retryCount++;
            }
        }
        if (retryCount == MAX_RETRY_COUNT) {
            // 达到最大重试次数，处理失败，记录日志等操作
        }
    }
}

在上面的代码中，我们创建了一个名为RetryConsumer的消费者，它使用了自定义的重试逻辑。主要思路是在消费消息时，如果处理失败，就进行重试，最多重试MAX_RETRY_COUNT次。如果达到最大重试次数，就认为处理失败，并可以进行相应的操作，如记录日志等。

另外，在重新平衡之前，我们使用了consumer.commitSync()来提交偏移量，以确保在重新平衡之后，消费者可以从最新的偏移量开始消费。在重新平衡之后，我们使用了consumer.seekToEnd(partitions)来重置消费者的偏移量，以从最新的消息开始消费。

注意：上述代码示例仅为演示目的，并没有处理一些其他的边界情况，如异常处理、线程安全等。在实际使用中，需要根据具体需求进行相应的改进和优化。