Kafka分区机制介绍与示例

来自： http://lxw1234.com/archives/2015/10/538.htm

Kafka中可以将Topic从物理上划分成一个或多个分区（Partition），每个分区在物理上对应一个文件夹， 以”topicName_partitionIndex”的命名方式命名，该文件夹下存储这个分区的所有消息(.log)和索引文件(.index)，这 使得Kafka的吞吐率可以水平扩展。

生产者在生产数据的时候，可以为每条消息指定Key，这样消息被发送到broker时，会根据分区规则选择被存储到哪一个分区中，如果分区规则设置 的合理，那么所有的消息将会被均匀的分布到不同的分区中，这样就实现了负载均衡和水平扩展。另外，在消费者端，同一个消费组可以多线程并发的从多个分区中 同时消费数据（后续将介绍这块）。

上面所说的分区规则，是实现了kafka.producer.Partitioner接口的类，可以自定义。比如，下面的代码 SimplePartitioner中，将消息的key做了hashcode，然后和分区数（numPartitions）做模运算，使得每一个key都 可以分布到一个分区中

package com.lxw1234.kafka;   import kafka.producer.Partitioner;  import kafka.utils.VerifiableProperties;   public class SimplePartitioner implements Partitioner {    public SimplePartitioner (VerifiableProperties props) { }    @Override public int partition(Object key, int numPartitions) {    int partition = 0;    String k = (String)key;    partition = Math.abs(k.hashCode()) % numPartitions;    return partition;   }   }

在创建Topic时候可以使用–partitions <numPartitions>指定分区数。也可以在server.properties配置文件中配置参数num.partitions来指定默认的分区数。

但有一点需要注意，为Topic创建分区时，分区数最好是broker数量的整数倍，这样才能是一个Topic的分区均匀的分布在整个Kafka集群中，假设我的Kafka集群由4个broker组成，以下图为例：

现在创建一个topic “lxw1234”，为该topic指定4个分区，那么这4个分区将会在每个broker上各分布一个：

./kafka-topics.sh   --create   --zookeeper zk1:2181,zk2:2181,zk3:2181   --replication-factor 1  --partitions 4   --topic lxw1234

这样所有的分区就均匀分布在集群中，如果创建topic时候指定了3个分区，那么就有一个broker上没有该topic的分区。

带分区规则的生产者

现在用一个生产者示例（PartitionerProducer），向Topic lxw1234中发送消息。该生产者使用的分区规则，就是上面的SimplePartitioner。从0-10一共11条消息，每条消息的key 为”key”+index，消息内容为”key”+index+”–value”+index。比如：key0–value0、key1– value1、、、key10–value10。

package com.lxw1234.kafka;  import java.util.Properties;  import kafka.javaapi.producer.Producer;  import kafka.producer.KeyedMessage;  import kafka.producer.ProducerConfig;   public class PartitionerProducer {   public static void main(String[] args)   { Properties props = new Properties();    props.put("serializer.class", "kafka.serializer.StringEncoder");    props.put("metadata.broker.list", "127.0.0.17:9091,127.0.0.17:9092,127.0.0.102:9091,127.0.0.102:9092");     props.put("partitioner.class", "com.lxw1234.kafka.SimplePartitioner");  Producer<String, String> producer = new Producer<String, String>(new ProducerConfig(props));        String topic = "lxw1234";            for(int i=0; i<=10; i++) {              String k = "key" + i;             String v = k + "--value" + i;                producer.send(new KeyedMessage<String, String>(topic,k,v));               }                   producer.close();                 }                }

理论上来说，生产者在发送消息的时候，会按照SimplePartitioner的规则，将key0做hashcode，然后和分区数（4）做模运算，得到分区索引：

hashcode(”key0”) % 4 = 1

hashcode(”key1”) % 4 = 2

hashcode(”key2”) % 4 = 3

hashcode(”key3”) % 4 = 0

         ……

对应的消息将会被发送至相应的分区中。

统计各分区消息的消费者

下面的消费者代码用来验证，在消费数据时，打印出消息所在的分区及消息内容：

package com.lxw1234.kafka;   import java.util.HashMap;  import java.util.List;  import java.util.Map;  import java.util.Properties;   import kafka.consumer.Consumer;  import kafka.consumer.ConsumerConfig;  import kafka.consumer.ConsumerIterator;  import kafka.consumer.KafkaStream;  import kafka.javaapi.consumer.ConsumerConnector;  import kafka.message.MessageAndMetadata;   public class MyConsumer {   public static void main(String[] args)   { String topic = "lxw1234";     ConsumerConnector consumer = Consumer.createJavaConsumerConnector(createConsumerConfig());     Map<String, Integer> topicCountMap = new HashMap<String, Integer>();      topicCountMap.put(topic, new Integer(1));      Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);      KafkaStream<byte[], byte[]> stream =  consumerMap.get(topic).get(0);       ConsumerIterator<byte[], byte[]> it = stream.iterator();     while(it.hasNext()) {          MessageAndMetadata<byte[], byte[]> mam = it.next();    System.out.println("consume: Partition [" + mam.partition() + "] Message: [" + new String(mam.message()) + "] ..");    private static ConsumerConfig createConsumerConfig() {      Properties props = new Properties();       props.put("group.id","group1");       props.put("zookeeper.connect","127.0.0.132:2181,127.0.0.133:2182,127.0.0.134:2183");      props.put("zookeeper.session.timeout.ms", "400")           props.put("zookeeper.sync.time.ms", "200");           props.put("auto.commit.interval.ms", "1000");           props.put("auto.offset.reset", "smallest");           return new ConsumerConfig(props);         }      }         Map<String, Integer> topicCountMap = new HashMap<String, Integer>();        topicCountMap.put(topic, new Integer(1));        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);        KafkaStream<byte[], byte[]> stream =  consumerMap.get(topic).get(0);        ConsumerIterator<byte[], byte[]> it = stream.iterator();           while(it.hasNext())           {                MessageAndMetadata<byte[], byte[]> mam = it.next();                System.out.println("consume: Partition [" + mam.partition() + "] Message: [" + new String(mam.message()) + "] ..");     }        }            private static ConsumerConfig createConsumerConfig() {           Properties props = new Properties();           props.put("group.id","group1");          props.put("zookeeper.connect","127.0.0.132:2181,127.0.0.133:2182,127.0.0.134:2183");           props.put("zookeeper.session.timeout.ms", "400");           props.put("zookeeper.sync.time.ms", "200");              props.put("auto.commit.interval.ms", "1000");           props.put("auto.offset.reset", "smallest");              return new ConsumerConfig(props);              }             }