Spring 卡夫卡分区
Spring Kafka Partitioning
以下两个代码片段发布消息的行为有何不同?
方法 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);
方法二
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 partitions/consumers.
之间平分
但是您的代码中有答案。
第一个与 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
的所有消息都将发送到同一分区。否则以循环方式放置到主题中。
以下两个代码片段发布消息的行为有何不同?
方法 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);
方法二
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 partitions/consumers.
之间平分但是您的代码中有答案。
第一个与 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
的所有消息都将发送到同一分区。否则以循环方式放置到主题中。