春季卡夫卡分区答案

【问题标题】：Spring Kafka Partitioning春季卡夫卡分区
【发布时间】：2018-01-15 07:18:52
【问题描述】：

下面两个代码sn-ps发布消息的行为有什么不同？

方法 1

Message<String> message = MessageBuilder.withPayload("testmsg")
        .setHeader(KafkaHeaders.MESSAGE_KEY, "key").setHeader(KafkaHeaders.TOPIC, "test").build();

ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(message);

方法 2

ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("test", "testmsg");

主题配置：

$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
Topic:test   PartitionCount:3    ReplicationFactor:1 Configs:
Topic: test  Partition: 0    Leader: 0   Replicas: 0 Isr: 0
Topic: test  Partition: 1    Leader: 0   Replicas: 0 Isr: 0
Topic: test  Partition: 2    Leader: 0   Replicas: 0 Isr: 0

观察：

如果有 3 个消费者，每个分区一个；方法 1 导致单个消费者从单个分区消费所有消息。采用方法 2；消费在 3 个分区/消费者之间平均分配。

【问题讨论】：

标签： apache-kafka kafka-consumer-api kafka-producer-api spring-kafka

【解决方案1】：

但是您的代码中有答案。第一个与topic 一起提供messageKey。

如果没有明确指定，messageKey 确实用于确定目标分区：

/**
 * computes partition for given record.
 * if the record has partition returns the value otherwise
 * calls configured partitioner class to compute the partition.
 */
private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
    Integer partition = record.partition();
    return partition != null ?
            partition :
            partitioner.partition(
                    record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
}

DefaultPartitioner 在哪里这样做：

List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
if (keyBytes == null) {
    int nextValue = nextValue(topic);
        ...
} else {
   // hash the keyBytes to choose a partition
   return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}

因此，所有具有相同key 的消息都被发送到同一个分区。否则，它们将被放置到主题循环方式中。

【讨论】：