在消息直接发布到 Kafka 分区的情况下进行分区复制

Question

据我了解，生产者和消费者对包含领导分区的领导代理进行写入和读取。记录写入leader分区后，记录会被复制到follower分区。

我最近读到 Producer 可以显式写入特定分区。这是否可能，如果可以，复制是否会发生在其他分区，因为它可能不是领导分区。

Answer 1

是的，您可以显式写入特定分区。 每个分区都有一个领导节点，其余节点是追随者。 当您写入一个分区时，它会被写入该分区的领导节点，并且跟随节点复制该数据。

例如，如果您启动一个包含 3 个代理节点 (broker1, broker2, broker3) 的集群，假设您创建了一个包含 3 个分区 (0, 1, 2) 的主题并将其配置为具有 2 个副本。

在这种情况下，每个代理都将成为一个分区的领导者，而其他两个分区的副本。比方说：

partition 0 has leader as broker1 and replicas as broker2 and broker3
partition 1 has leader as broker2 and replicas as broker1 and broker3
partition 2 has leader as broker3 and replicas as broker1 and broker2

当您写入分区 0 时，它会写入它的领导节点 broker1，然后复制到其他节点，其他分区也是如此。

即使 broker2 是分区 1 的领导者，它也是分区 0 和 2 的追随者，并且任何需要从分区 0 或 2 复制的消息都将写入 broker2，因为它只是分区 1 的领导者。

Answer 2

“我最近读到 Producer 可以显式写入特定分区。这可能吗[...]”

是的，这是真的。您可以通过实现Partitioner 来定义自定义分区器，并在KafkaProducer 配置中声明它partitioner.class

以下是 Java 中的自定义分区器示例：

public class MyPartitioner implements Partitioner {
  public void configure(Map<String, ?> configs) {}
  public void close() {}

  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();

    if ((keyBytes == null) || (!(key instanceOf String)))
      throw new InvalidRecordException("Record did not have a string Key");

    if (((String) key).equals("myKey"))
       return 0; // This key will always go to Partition 0

    // Other records will go to the rest of the Partitions using a hashing function
    return (Math.abs(Utils.murmur2(keyBytes)) % (numPartitions - 1)) + 1;
  }
}

“[...] 如果是这样，复制是否会发生在其他分区，因为它可能不是领导分区？”

您正在将分区与副本混为一谈。这是两个独立的概念，因为分区将主题的数据分布到多个位置以实现并行性，而复制实际上是跨集群复制数据以获得更高的持久性。

使用自定义分区器仍将确保您的 KafkaProducer 只会与分区领导者通信并将数据写入分区领导者。

请记住，主题中的每个分区都有一个分区领导者。您拥有的分区领导者数量与主题中的分区数量一样多。