面试题：消息队列 Kafka 在大规模分布式系统中，面对网络波动等异常情况，如何保证端到端的消息顺序性？

1. Kafka 集群配置

分区设计：
- 为了保障消息顺序性，将具有相同顺序要求的消息发送到同一个分区。例如，对于某个用户的所有操作消息，根据用户 ID 进行分区，确保来自同一用户的消息都在一个分区内。在 Kafka 中，可以通过自定义分区器来实现这一点。假设消息包含用户 ID 字段，自定义分区器代码示例（以 Java 为例）：

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.utils.Utils;
import java.util.Map;

public class UserIdPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        String userId = (String) key;
        int numPartitions = cluster.partitionsForTopic(topic).size();
        return Math.abs(userId.hashCode()) % numPartitions;
    }

    @Override
    public void close() { }

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

- 在生产者配置中指定该分区器：

Properties props = new Properties();
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, UserIdPartitioner.class.getName());

副本因子设置：
- 适当提高副本因子，比如设置为 3 或更高，以增强数据的可靠性。在节点故障时，有更多的副本可以接替工作，减少消息丢失的风险。例如，创建主题时设置副本因子：

kafka-topics.sh --create --topic my_topic --partitions 10 --replication-factor 3 --bootstrap-server kafka:9092

2. 生产者可靠性设置

acks 参数：
- 设置 acks = all（或 acks = -1），表示生产者需要等待所有副本都确认收到消息后才认为消息发送成功。这样可以确保即使部分副本所在节点出现故障，消息也不会丢失。生产者配置示例：

props.put(ProducerConfig.ACKS_CONFIG, "all");

retries 参数：
- 配置合理的 retries 值，比如 retries = 3。当网络波动导致消息发送失败时，生产者会自动重试。同时，结合 retry.backoff.ms 参数设置重试间隔，例如 retry.backoff.ms = 500，即每次重试间隔 500 毫秒，避免过于频繁的重试。

props.put(ProducerConfig.RETRIES_CONFIG, 3);
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 500);

3. 消息传输过程

事务支持：
- Kafka 从 0.11.0.0 版本开始支持事务。如果业务场景需要保证一组消息的原子性，即要么所有消息都成功发送，要么都失败，可以使用事务。例如，在一个电商系统中，下单消息和库存更新消息需要作为一个事务处理。
- 生产者端代码示例（以 Java 为例）：

Producer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders_topic", "order_id_1", "order_message"));
    producer.send(new ProducerRecord<>("inventory_topic", "product_id_1", "inventory_update_message"));
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    producer.abortTransaction();
} catch (KafkaException e) {
    producer.abortTransaction();
}

4. 消费者可靠性设置

消费组设置：
- 确保每个分区只有一个消费者实例进行消费，避免多个消费者并发消费同一分区导致消息顺序混乱。可以通过合理设置消费组的消费者数量来实现，一般来说，消费组内消费者数量不超过分区数量。
offset 管理：
- 采用手动提交 offset 的方式，确保消息被成功处理后再提交 offset。例如，在处理完一批消息后，手动提交 offset：

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    // 处理消息
    System.out.println("Received message: " + record.value());
}
consumer.commitSync();

单线程消费：
- 在消费者端使用单线程消费，这样可以保证消息按照分区中的顺序依次被处理。例如，在 Java 中可以创建一个线程池，线程池大小为 1，由这个单线程来处理所有的消费逻辑：

ExecutorService executor = Executors.newFixedThreadPool(1);
executor.submit(() -> {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // 处理消息
            System.out.println("Received message: " + record.value());
        }
        consumer.commitSync();
    }
});

星途面试题库

面试题：消息队列 Kafka 在大规模分布式系统中，面对网络波动等异常情况，如何保证端到端的消息顺序性？

知识考点

面试题答案

1. Kafka 集群配置

2. 生产者可靠性设置

3. 消息传输过程

4. 消费者可靠性设置