JAVA Kafka 消息乱序？生产者分区策略与消费组同步机制讲解

各位朋友，大家好！今天我们来聊聊 Kafka 中一个常见但又容易令人困惑的问题：消息乱序。 Kafka 作为一种高吞吐、分布式的消息队列，在很多场景下都扮演着重要的角色。然而，在某些情况下，我们可能会发现消费者接收到的消息顺序与生产者发送的顺序不一致，也就是出现了乱序。为什么会发生乱序？Kafka 真的保证不了消息的顺序性吗？本次讲座，我们将深入探讨 Kafka 的生产者分区策略和消费组同步机制，从根本上理解消息乱序的原因，并学习如何避免或解决这个问题。

一、Kafka 消息顺序性的保障范围

首先，我们要明确一点：Kafka 并非在所有情况下都能保证全局的消息顺序性。 Kafka 保证的是分区内的消息顺序性。也就是说，对于同一个分区而言，消息的写入顺序和读取顺序是一致的。

为什么 Kafka 要做这样的设计？这是出于性能的考虑。全局顺序性意味着所有消息必须经过同一个节点进行排序和写入，这会极大地限制 Kafka 的吞吐量和并发能力。分区机制允许 Kafka 将消息分散到多个节点上进行处理，从而实现更高的性能。

二、生产者分区策略：消息进入哪个分区？

生产者在发送消息时，需要决定将消息发送到哪个分区。这个决策过程由分区策略决定。 Kafka 提供了多种内置的分区策略，也可以自定义分区策略。

默认分区策略（DefaultPartitioner）：

如果消息指定了 key，则使用 key 的哈希值对分区数取模，将消息发送到对应的分区。
如果消息没有指定 key，则采用轮询（Round-Robin）的方式，将消息均匀地发送到各个分区。从 Kafka 2.4 版本开始，默认采用黏性轮询（Sticky Partitioning）策略，在一次批处理中，尽可能将消息发送到同一个分区，以提高效率。

代码示例：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;

public class KafkaProducerExample {

    public static void main(String[] args) throws Exception {

        String topicName = "my-topic";

        // Kafka Broker 地址
        String bootstrapServers = "localhost:9092";

        // 配置生产者属性
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 创建 Kafka 生产者
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        try {
            // 发送带 key 的消息 (key 会影响分区)
            ProducerRecord<String, String> record1 = new ProducerRecord<>(topicName, "key1", "message1");
            producer.send(record1);

            ProducerRecord<String, String> record2 = new ProducerRecord<>(topicName, "key1", "message2");
            producer.send(record2);

            // 发送不带 key 的消息 (轮询或黏性轮询)
            ProducerRecord<String, String> record3 = new ProducerRecord<>(topicName, null, "message3");
            producer.send(record3);

            ProducerRecord<String, String> record4 = new ProducerRecord<>(topicName, null, "message4");
            producer.send(record4);

            producer.flush(); // 确保所有消息都已发送
        } finally {
            producer.close();
        }
    }
}

解释：

record1 和 record2 拥有相同的 key "key1"，因此它们会被发送到同一个分区。在这个分区内，message1 和 message2 的顺序会被保证。
record3 和 record4 没有指定 key，它们会被轮询发送到不同的分区 (或在黏性分区下，在一段时间内发送到同一分区，直到该分区被填满)。因此，message3 和 message4 的顺序无法保证。

自定义分区策略：

你可以通过实现 org.apache.kafka.clients.producer.Partitioner 接口来定义自己的分区策略。这允许你根据业务需求，实现更复杂的分区逻辑。

代码示例：

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

public class CustomPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        Integer partitionCount = cluster.partitionsForTopic(topic).size();
        if (key == null) {
            // 如果 key 为空，随机选择一个分区
            return (int) (Math.random() * partitionCount);
        }

        // 根据 key 的哈希值选择分区
        return Math.abs(key.hashCode()) % partitionCount;
    }

    @Override
    public void close() {
        // 可选：释放资源
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 可选：配置分区器
    }
}

配置生产者使用自定义分区器：

props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "CustomPartitioner");

解释：

partition() 方法是分区策略的核心。它接收 topic、key、value 等信息，并返回一个分区 ID。
在 CustomPartitioner 中，如果 key 为空，则随机选择一个分区；否则，根据 key 的哈希值选择分区。

消息 Key 的重要性：

通过上面的例子，我们可以看出，消息的 key 在分区策略中扮演着重要的角色。如果需要保证具有某种关联性的消息的顺序，那么这些消息应该使用相同的 key。

举例:

假设我们需要记录用户订单的状态变更，并保证同一个用户的订单状态变更消息按照时间顺序消费。那么，我们可以将用户 ID 作为 key，将订单状态变更消息作为 value。这样，所有关于同一个用户的订单状态变更消息都会被发送到同一个分区，从而保证顺序性。

三、消费组同步机制：如何消费消息？

消费者通过消费组（Consumer Group）来消费 Kafka 中的消息。消费组允许多个消费者并行地消费同一个主题的不同分区，从而提高消费速度。 Kafka 通过消费组同步机制来协调消费组内的消费者，确保每个分区只被一个消费者消费。

消费组的组成：

一个消费组由一个或多个消费者组成。每个消费者都有一个唯一的 ID。
分区分配策略：

Kafka 会根据分区分配策略，将主题的分区分配给消费组内的消费者。常见的分区分配策略有：
- RangeAssignor： 将分区按照顺序分配给消费者。例如，如果一个主题有 5 个分区，一个消费组有 2 个消费者，那么第一个消费者会被分配到分区 0、1 和 2，第二个消费者会被分配到分区 3 和 4。
- RoundRobinAssignor： 轮询地将分区分配给消费者。例如，如果一个主题有 5 个分区，一个消费组有 2 个消费者，那么第一个消费者会被分配到分区 0、2 和 4，第二个消费者会被分配到分区 1 和 3。
- StickyAssignor： 尝试将分区尽可能地分配给相同的消费者，以减少消费者的切换开销。默认策略。
- CooperativeStickyAssignor： 合作式黏性分配策略，允许在消费者加入或离开消费组时，进行增量式的分区分配，避免全局重新分配，减少消费中断时间。
消费组同步过程：

当消费组内的消费者发生变化时（例如，有新的消费者加入，或者有消费者离开），Kafka 会触发消费组的同步过程。

同步过程主要包括以下几个步骤：
- Leader选举： 消费组内的所有消费者会竞争成为 Leader。Leader 负责制定分区分配方案。
- 分区分配： Leader 根据分区分配策略，将主题的分区分配给消费组内的消费者。
- 同步元数据： Leader 将分区分配方案发送给消费组内的所有消费者。
- 消费者启动消费： 消费者根据分区分配方案，开始消费自己负责的分区。

消费位移（Offset）管理：

每个消费者会记录自己消费到的消息的位移（Offset）。位移是指消息在分区中的位置。消费者会将位移信息提交给 Kafka，以便在消费者重启或重新加入消费组时，能够从上次消费的位置继续消费。 Kafka 提供了多种位移提交方式：

自动提交： 消费者会自动定期提交位移。
手动提交： 消费者可以手动控制位移的提交时机。

代码示例：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {

    public static void main(String[] args) {
        String topicName = "my-topic";
        String groupId = "my-group";

        // 配置消费者属性
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从最早的消息开始消费

        // 创建 Kafka 消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList(topicName));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, key = %s, value = %s, partition = %d%n", record.offset(), record.key(), record.value(), record.partition());
                }
                //自动提交位移，也可以手动提交
                //consumer.commitSync();
            }
        } finally {
            consumer.close();
        }
    }
}

解释：

consumer.subscribe(Collections.singletonList(topicName)) 订阅主题。
consumer.poll(Duration.ofMillis(100)) 从 Kafka 拉取消息。
consumer.commitSync() 手动同步提交位移。不调用该方法，默认会自动提交。
四、消息乱序的常见原因与解决方案

现在，我们来总结一下导致 Kafka 消息乱序的常见原因，并给出相应的解决方案：

原因	解决方案
生产者发送到不同的分区	确保具有关联性的消息使用相同的 `key`，以便它们被发送到同一个分区。
生产者重试机制导致消息重发	1. 设置 `retries` 为 0，禁用重试。但这可能会导致消息丢失。 2. 使用幂等性生产者，确保消息只被发送一次。开启 `enable.idempotence` 为 `true`。 3. 在消费者端进行去重处理。
消费者并行消费多个分区	如果需要保证全局顺序性，则将主题的分区数设置为 1，并使用一个消费者消费该分区。但这会降低消费速度。
消费者处理消息失败并重试	如果消费者处理消息失败并重试，可能会导致消息的消费顺序与发送顺序不一致。确保消费者具备处理消息失败的能力，例如，将失败的消息放入死信队列，稍后进行处理。
消费者手动提交位移的逻辑错误	如果消费者手动提交位移的逻辑错误，可能会导致消息被重复消费或遗漏消费。仔细检查手动提交位移的逻辑，确保位移提交的正确性。
主题分区数大于消费者数量，导致部分消费者空闲，分区分配不均	增加消费者数量，使得每个消费者都能够分配到分区。或者，减少主题的分区数量，使其与消费者数量相匹配。如果使用自动分区分配，则 Kafka 会自动将分区分配给消费者。如果使用手动分区分配，则需要手动调整分区分配方案。

五、幂等性生产者与事务性生产者

为了解决消息重复发送的问题，Kafka 提供了幂等性生产者和事务性生产者两种机制。

幂等性生产者：

通过为每个生产者分配一个唯一的 Producer ID (PID)，并为每个消息分配一个序列号，Kafka 能够识别重复发送的消息，并只保留第一条消息。

开启幂等性：
```
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
```
注意：
- 开启幂等性后，retries 必须大于 0。
- max.in.flight.requests.per.connection 必须小于等于 5。

事务性生产者：

事务性生产者允许将多个消息作为一个原子单元发送。要么所有消息都成功发送，要么所有消息都失败。

使用事务性生产者：

props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>(topicName, "key1", "message1"));
    producer.send(new ProducerRecord<>(topicName, "key2", "message2"));
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
} finally {
    producer.close();
}

注意：

需要设置 transactional.id。
需要在消费者端开启事务支持，设置 isolation.level 为 read_committed。

六、总结与应用实践

Kafka 保证分区内的消息顺序性，但全局顺序性需要额外的考虑。通过合理的分区策略和消费组同步机制，我们可以有效地控制消息的顺序和消费方式。在实际应用中，根据业务需求选择合适的分区策略和消费组配置，可以避免消息乱序问题，并充分发挥 Kafka 的性能优势。

七、核心要点回顾

Kafka 的顺序性保障范围是分区内，而不是全局。生产者通过分区策略决定消息进入哪个分区，相同 Key 的消息会被发送到同一分区。消费组同步机制协调消费者，确保每个分区只被一个消费者消费。

八、常见乱序原因与预防

消息乱序的常见原因包括生产者发送到不同分区、重试机制导致重发、消费者并行消费多个分区等。可以通过使用相同的 Key、幂等性生产者、事务性生产者等方式来避免乱序。

九、最佳实践建议

在设计 Kafka 应用时，要充分考虑消息的顺序性需求，选择合适的分区策略和消费组配置。同时，要注意处理消息失败的情况，并监控 Kafka 的运行状态，及时发现和解决问题。

JAVA Kafka 消息乱序？生产者分区策略与消费组同步机制讲解

发表回复 取消回复

发表回复取消回复