滴滴二面之Kafka如何读写副本消息的攻略

Kafka 是一种分布式消息系统，消息被分为多个分区存储在多个 broker 中。副本是为了在发生故障时提供消息持久性和可靠性所增加的。在 Kafka 中，每个分区都会有多个副本，其中一个作为主副本，其他副本作为从副本，主副本负责进行读写操作，而从副本只需要对主副本的写操作进行复制，从而保证数据的可靠性。

读副本消息

在 Kafka 中，不仅可以读取主副本中的消息，还可以读取从副本中的消息。如果要读取从副本中的消息，需要更改 Consumer 的属性。

一般情况下，我们使用以下代码来创建消费者并读取消息：

public class ConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "my-group");
        props.setProperty("enable.auto.commit", "true");
        props.setProperty("auto.commit.interval.ms", "1000");
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record);
            }
        }
    }
}

如果要读取从副本中的消息，只需要将 fetch.min.bytes 属性设置为 1 即可：

public class ConsumerExampleWithReplica {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "my-group");
        props.setProperty("enable.auto.commit", "true");
        props.setProperty("auto.commit.interval.ms", "1000");
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("fetch.min.bytes", "1");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record);
            }
        }
    }
}

写副本消息

当向 Kafka 中写入消息时，可以将消息发送到主副本或者多个副本。如果将消息发送到主副本，则此消息只会保存在主副本中。如果将消息发送到多个副本，则每个副本都会保存一份拷贝。这种情况下，写入的速度会变慢，但可以提高消息的可靠性。

以下是向多个副本中写入消息的示例代码：

public class ProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

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

        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                exception.printStackTrace();
            } else {
                System.out.printf("Message sent to partition %d with offset %d\n", metadata.partition(), metadata.offset());
            }
        });

        producer.close();
    }
}