bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic my-topic --partitions 2 --replication-factor 1

[root@localhost kafka_2.11-2.0.0]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic abc
Topic:abc       PartitionCount:2        ReplicationFactor:1     Configs:
        Topic: abc      Partition: 0    Leader: 0       Replicas: 0     Isr: 0
        Topic: abc      Partition: 1    Leader: 0

/**
 * Compute the partition for the given record.
 *
 * @param topic The topic name
 *  key The key to partition on (or null if no key)
 *  keyBytes serialized key to partition on (or null if no key)
 *  value The value to partition on or null
 *  valueBytes serialized value to partition on or null
 *  cluster The current cluster Metadata
 */
public int partition(String topic,Object key,byte[] keyBytes,Object value,1)">byte[] valueBytes,Cluster cluster) {
    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
    int numPartitions = partitions.size();
    if (keyBytes == null) {
        int nextValue = nextValue(topic);
        List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
        if (availablePartitions.size() > 0) {
            int part = Utils.toPositive(nextValue) % availablePartitions.size();
            return availablePartitions.get(part).partition();
        } else {
            // no partitions are available,give a non-available partition
            return Utils.toPositive(nextValue) % numPartitions;
        }
    }  {
         hash the keyBytes to choose a partition
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }
}

 * The range assignor works on a per-topic basis. For each topic,we lay out the available partitions in numeric order
 * and the consumers in lexicographic order. We then divide the number of partitions by the total number of
 * consumers to determine the number of partitions to assign to each consumer. If it does not evenly
 * divide,then the first few consumers will have one extra partition.
 *
 * For example,suppose there are two consumers C0 and C1,two topics t0 and t1,and each topic has 3 partitions,* resulting in partitions t0p0,t0p1,t0p2,t1p0,t1p1,and t1p2.
 *
 * The assignment will be:
 * C0: [t0p0,t1p1]
 * C1: [t0p2,t1p2]
 */

public Map<String,List<TopicPartition>> assign(Map<String,Integer> partitionsPerTopic,Map<String,Subscription> subscriptions) {
        主题与消费者的映射                                                            
    Map<String,List<String>> consumersPerTopic = consumersPerTopic(subscriptions);
    Map<String,List<TopicPartition>> assignment = new HashMap<>();
    for (String memberId : subscriptions.keySet())
        assignment.put(memberId,new ArrayList<TopicPartition>());

    for (Map.Entry<String,List<String>> topicEntry : consumersPerTopic.entrySet()) {
        String topic = topicEntry.getKey();        主题
        List<String> consumersForTopic = topicEntry.getValue();        消费者列表

            partitionsPerTopic表示主题和分区数的映射
            获取主题下有多少个分区
        Integer numPartitionsForTopic = partitionsPerTopic.get(topic);
        if (numPartitionsForTopic == )
            continue;

            消费者按字典序排序
        Collections.sort(consumersForTopic);

            分区数量除以消费者数量
        int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size();
            取模，余数就是额外的分区
        int consumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size();

        List<TopicPartition> partitions = AbstractPartitionAssignor.partitions(topic,numPartitionsForTopic);
        for (int i = 0,n = consumersForTopic.size(); i < n; i++int start = numPartitionsPerConsumer * i + Math.min(i,consumersWithExtraPartition);
            int length = numPartitionsPerConsumer + (i + 1 > consumersWithExtraPartition ? 0 : 1);
                分配分区
            assignment.get(consumersForTopic.get(i)).addAll(partitions.subList(start,start + length));
        }
    }
     assignment;
}

 * The round robin assignor lays out all the available partitions and all the available consumers. It
 * then proceeds to do a round robin assignment from partition to consumer. If the subscriptions of all consumer
 * instances are identical,then the partitions will be uniformly distributed. (i.e.,the partition ownership counts
 * will be within a delta of exactly one across all consumers.)
 *
 * For example,t1p1]
 * C1: [t0p1,t1p2]
 *
 * When subscriptions differ across consumer instances,the assignment process still considers each
 * consumer instance in round robin fashion but skips over an instance if it is not subscribed to
 * the topic. Unlike the case when subscriptions are identical,this can result in imbalanced
 * assignments. For example,we have three consumers C0,C1,C2,and three topics t0,t1,t2,* with 1,2,and 3 partitions,respectively. Therefore,the partitions are t0p0,t2p0,* t2p1,t2p2. C0 is subscribed to t0; C1 is subscribed to t0,t1; and C2 is subscribed to t0,t2.
 *
 * Tha assignment will be:
 * C0: [t0p0]
 * C1: [t1p0]
 * C2: [t1p1,t2p1,t2p2]
 */

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    modelVersion>4.0.0</>

    groupId>com.cjs.exampleartifactId>kafka-demoversion>0.0.1-SNAPSHOTpackaging>jarnamedescription></parent>
        >org.springframework.boot>spring-boot-starter-parent>2.0.5.RELEASErelativePath/> <!-- lookup parent from repository -->
    propertiesproject.build.sourceEncoding>UTF-8project.reporting.outputEncodingjava.version>1.8dependenciesdependency>
            >spring-boot-starter-web>org.springframework.kafka>spring-kafka>

        >spring-boot-starter-testscope>testbuildpluginsplugin>
                >spring-boot-maven-plugin>

project>

package com.cjs.kafka.producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

class HelloProducer {

    static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers","192.168.1.133:9092");
        props.put("acks","all");
        props.put("retries",0);
        props.put("batch.size",16384);
        props.put("linger.ms",1);
        props.put("buffer.memory",33554432);
        props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer",1)">);

        Producer<String,String> producer = new KafkaProducer<String,String>(props);
        int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<String,String>("abc",Integer.toString(i),Integer.toString(i)),1)"> Callback() {
                @Override
                 onCompletion(RecordMetadata recordMetadata,Exception e) {
                    if (null != e) {
                        e.printStackTrace();
                    } {
                        System.out.println("callback: " + recordMetadata.topic() + " " + recordMetadata.partition() + " " + recordMetadata.offset());
                    }
                }
            });
        }
        producer.close();

    }
}

 com.cjs.kafka.consumer;

 org.apache.kafka.clients.consumer.ConsumerRecord;
 org.apache.kafka.clients.consumer.ConsumerRecords;
 org.apache.kafka.clients.consumer.KafkaConsumer;

 java.util.Arrays;
 HelloConsumer {

     main(String[] args) {
        Properties props = );
        props.put("group.id","test");
        props.put("enable.auto.commit","true");
        props.put("auto.commit.interval.ms","1000");
        props.put("partition.assignment.strategy","org.apache.kafka.clients.consumer.RoundRobinAssignor");
        props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",1)">);
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String,1)">(props);
        consumer.subscribe(Arrays.asList("foo","bar","abc"));
        while (true) {
            ConsumerRecords<String,String> records = consumer.poll(100for (ConsumerRecord<String,1)"> record : records) {
                System.out.printf("partition = %s,offset = %d,key = %s,value = %s%n",record.partition(),record.offset(),record.key(),record.value());
            }
        }
    }
}