package cn.hadoop.hdfs.kafka.partition;

import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

/**
 * @Date Nov 3, 2016
 *
 * @Author dengjie
 *
 * @Note 先 Hash 再取模，得到分区索引
 */
public class CustomerPartitioner implements Partitioner {

    public CustomerPartitioner(VerifiableProperties props) {
    }

    public int partition(Object key, int numPartitions) {
        int partition = 0;
        String k = (String) key;
        partition = Math.abs(k.hashCode()) % numPartitions;
        return partition;
    }

}

./kafka-topics.sh --create --zookeeper k1:2181,k2:2181,k3:2181 --replication-factor 3 --partitions 6 --topic ke_test

package cn.hadoop.hdfs.kafka.partition;

import java.util.List;
import java.util.Properties;

import cn.hadoop.hdfs.kafka.partition.data.FileRead;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
 * @Date Nov 3, 2016
 *
 * @Author dengjie
 *
 * @Note 按照先 Hash 再取模的规则，进行分区入库
 */
public class PartitionerProducer {
    public static void main(String[] args) {
        producerData();
    }

    private static void producerData() {
        Properties props = new Properties();
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("metadata.broker.list", "k1:9092,k2:9092,k3:9092");
        props.put("partitioner.class", "cn.hadoop.hdfs.kafka.partition.CustomerPartitioner");
        Producer<String, String> producer = new Producer<String, String>(new ProducerConfig(props));
        String topic = "ke_test";
        List<String> list = FileRead.readData();
        for (int i = 0; i < list.size(); i++) {
            String k = "key" + i;
            String v = new String(list.get(i));
            producer.send(new KeyedMessage<String, String>(topic, k, v));
            if (i == (list.size() - 1)) {
                return;
            }
        }
        producer.close();
    }
}

hashCode("key0") % 6 = 1
hashCode("key1") % 6 = 2
hashCode("key2") % 6 = 3
hashCode("key3") % 6 = 4
hashCode("key4") % 6 = 5
hashCode("key5") % 6 = 0
// ... 以此循环

package cn.hadoop.hdfs.kafka.partition;

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;

/**
 * @Date Nov 3, 2016
 *
 * @Author dengjie
 *
 * @Note 通过 Kafka 的消费者 API 验证分区入库的消息
 */
public class PartitionerConsumer {
    public static void main(String[] args) {
        String topic = "ke_test";
        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", "zk1:2181,zk2:2181,zk3:2181");
        props.put("zookeeper.session.timeout.ms", "40000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "smallest");
        return new ConsumerConfig(props);
    }
}