面试题：消息队列 Kafka 集群节点故障对消费者和生产者的影响及应对策略

Question

当 Kafka 集群出现节点故障时，对正在进行消息生产和消费的生产者与消费者会产生哪些具体影响？如何在代码层面进行优化，以降低这些影响并保障业务的连续性？

Answer 1

节点故障对生产者的影响

1. 消息发送失败：若故障节点是生产者正在向其发送消息的目标节点，消息可能无法成功发送，生产者会收到错误响应，比如 NetworkException 等。
2. 分区重分配：Kafka 为保证数据可用性会进行分区重分配，在这个过程中，生产者可能会因为找不到目标分区而发送失败。

节点故障对消费者的影响

1. 消费中断：如果故障节点包含消费者正在消费的分区副本，消费者会收到 OffsetOutOfRangeException 等错误，导致消费暂停，直到 Kafka 完成分区重新分配和副本同步。
2. 重新平衡：Kafka 会触发消费者组的重新平衡，这可能导致部分消费者在重新平衡期间短暂停止消费，造成消费延迟。

代码层面优化措施

生产者优化

1. 设置重试机制：在生产者代码中设置 retries 参数，例如在 Java 中使用 KafkaProducer 时：

Properties props = new Properties();
props.put("retries", 3);
// 其他配置
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

这样当消息发送失败时，生产者会自动重试指定次数。 2. 异步发送与回调处理：采用异步发送消息并设置回调函数，及时处理发送结果。例如：

producer.send(new ProducerRecord<>("topic", "key", "value"), (recordMetadata, e) -> {
    if (e != null) {
        // 处理发送失败情况
        System.err.println("发送消息失败: " + e.getMessage());
    } else {
        // 发送成功处理
        System.out.println("消息发送成功，偏移量: " + recordMetadata.offset());
    }
});

3. 合理设置分区策略：根据业务需求自定义分区策略，避免消息集中在少数节点，提高集群可用性。例如：

public class CustomPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // 自定义分区逻辑
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        return Math.abs(key.hashCode()) % numPartitions;
    }

    @Override
    public void close() {
        // 关闭资源
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 配置
    }
}

然后在生产者配置中指定分区器：

props.put("partitioner.class", "com.example.CustomPartitioner");

消费者优化

1. 自动偏移量提交改为手动提交：手动控制偏移量提交，避免在重新平衡时丢失已消费但未提交的偏移量。例如在 Java 中：

props.put("enable.auto.commit", "false");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 处理消息
        System.out.println("消费消息: " + record.value());
    }
    // 手动提交偏移量
    consumer.commitSync();
}

2. 设置消费者组再平衡监听器：在重新平衡前暂停消费，重新平衡后恢复消费。例如：

consumer.subscribe(Arrays.asList("topic"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 暂停消费
        consumer.pause(partitions);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 恢复消费
        consumer.resume(partitions);
    }
});

3. 合理设置消费者参数：如 max.poll.records 控制每次拉取的最大记录数，避免一次拉取过多数据导致处理时间过长，在重新平衡时出现问题。例如：

props.put("max.poll.records", 100);