标签：
kafka版本0.8.2.1
Java客户端版本0.9.0.0
为了更好的实现负载均衡和消息的顺序性，Kafka Producer可以通过分发策略发送给指定的Partition。Kafka保证在partition中的消息是有序的。Kafka Java客户端有默认的Partitioner。实现如下：
public int partition(ProducerRecord<byte[], byte[]> record, Cluster cluster) { List partitions = cluster.partitionsForTopic(record.topic()); int numPartitions = partitions.size(); if(record.partition() != null) { if(record.partition().intValue() >= 0 && record.partition().intValue() < numPartitions) { return record.partition().intValue(); } else { throw new IllegalArgumentException("Invalid partition given with record: " + record.partition() + " is not in the range [0..." + numPartitions + "]."); } } else if(record.key() == null) { int nextValue = this.counter.getAndIncrement(); List availablePartitions = cluster.availablePartitionsForTopic(record.topic()); if(availablePartitions.size() > 0) { int part = Utils.abs(nextValue) % availablePartitions.size(); return ((PartitionInfo)availablePartitions.get(part)).partition(); } else { return Utils.abs(nextValue) % numPartitions; } } else { return Utils.abs(Utils.murmur2((byte[])record.key())) % numPartitions; } }
从源码可以看出，首先获取topic的所有Patition，如果客户端不指定Patition，也没有指定Key的话，使用自增长的数字取余数的方式实现指定的Partition。这样Kafka将平均的向Partition中生产数据。测试代码如下：
Producer：
String topic = "haoxy1"; int i = 0; Properties props = new Properties(); props.put("bootstrap.servers", "10.23.22.237:9092,10.23.22.238:9092,10.23.22.239:9092"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props); System.out.println("partitions count " + producer.partitionsFor(topic)); while(true) { String msg = "test"+i++; ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topic, msg); producer.send(producerRecord); System.out.println("send " + msg); Thread.sleep(5000); }
Consumer：
String topic = "haoxy1"; Properties props = new Properties(); props.put("zookeeper.connect", "10.23.22.237:2181,10.23.22.238:2181,10.23.22.239:2181"); props.put("group.id", "cg.nick"); props.put("consumer.id", "c.nick"); Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); topicCountMap.put(topic, 3); ConsumerConfig consumerConfig = new ConsumerConfig(props); ConsumerConnector consumer = Consumer.createJavaConsumerConnector(consumerConfig); Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap); List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic); ExecutorService executor = Executors.newFixedThreadPool(3); for (final KafkaStream stream : streams) { executor.submit(new Runnable() { public void run() { ConsumerIterator<byte[], byte[]> it = stream.iterator(); while (it.hasNext()) { MessageAndMetadata<byte[], byte[]> mm = it.next(); System.out.println(String.format("partition = %s, offset = %d, key = %s, value = %s", mm.partition(), mm.offset(), mm.key(), new String(mm.message()))); } } }); }
从测试结果结果看出，是平均分配的：
partition = 1, offset = 416, key = null, value = test9 partition = 0, offset = 386, key = null, value = test10 partition = 2, offset = 454, key = null, value = test11 partition = 1, offset = 417, key = null, value = test12 partition = 0, offset = 387, key = null, value = test13 partition = 2, offset = 455, key = null, value = test14 partition = 1, offset = 418, key = null, value = test15 partition = 0, offset = 388, key = null, value = test16
如果想要控制发送的partition，则有两种方式，一种是指定partition，另一种就是根据Key自己写算法。继承Partitioner接口，实现其partition方法。并且配置启动参数
props.put("partitioner.class","TestPartitioner")。
比如需要实现
key=’aaa’ 的都进partition 0
key=’bbb’ 的都进partition 1
key=’bbb’ 的都进partition 2
public class TestPartitioner implements Partitioner { public int partition(String s, Object key, byte[] bytes, Object o1, byte[] bytes1, Cluster cluster) { if (key.toString().equals("aaa")) return 0; else if (key.toString().equals("bbb")) return 1; else if (key.toString().equals("ccc")) return 2; else return 0; } public void close() { } public void configure(Map<String, ?> map) { } }
测试结果：
partition = 0, offset = 438, key = aaa, value = test32 partition = 1, offset = 448, key = bbb, value = test33 partition = 2, offset = 486, key = ccc, value = test34 partition = 0, offset = 439, key = aaa, value = test35 partition = 1, offset = 449, key = bbb, value = test36 partition = 2, offset = 487, key = ccc, value = test37 partition = 0, offset = 440, key = aaa, value = test38 partition = 1, offset = 450, key = bbb, value = test39 partition = 2, offset = 488, key = ccc, value = test40 partition = 0, offset = 441, key = aaa, value = test41 partition = 1, offset = 451, key = bbb, value = test42 partition = 2, offset = 489, key = ccc, value = test43 partition = 0, offset = 442, key = aaa, value = test44
如果你使用的不是Java的客户端，是javaapi下面的Producer的话，自定义的分区类需要实现kafka.producer.Partitioner，并且有构造函数。
public class TestPartitioner implements Partitioner { public TestPartitioner (VerifiableProperties props) { } public int partition(Object o, int i) { if (o.toString().equals("aaa")) return 0; else if (o.toString().equals("bbb")) return 1; else if (o.toString().equals("ccc")) return 2; else return 0; } }
标签：
原文地址：http://www.cnblogs.com/haoxinyue/p/5743775.html
