Kafka消费者协调

Kafka消费者协调是指多个Kafka消费者协同工作以平衡负载和处理消息的过程。下面是一个示例，展示了如何使用Java编写的Kafka消费者协调代码：

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

public class KafkaConsumerExample {
    private static final String TOPIC_NAME = "my_topic";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String GROUP_ID = "my_consumer_group";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(TOPIC_NAME));

        try {
            while (true) {
                // 拉取消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    // 处理消息
                    System.out.printf("Received message: key = %s, value = %s, partition = %s, offset = %d%n",
                            record.key(), record.value(), record.partition(), record.offset());
                }

                // 手动提交偏移量
                consumer.commitSync();
            }
        } finally {
            consumer.close();
        }
    }
}

上述代码创建了一个Kafka消费者，并订阅了名为"my_topic"的主题。消费者从Kafka服务器拉取消息，并处理每条消息。在处理完消息后，消费者手动提交偏移量以确保消息被正确消费。