基本概念

topic

在kafka 中，topic是一个存储消息的逻辑概念，可以认为是一个消息的集合。每条消息发送到 kafka 集群的消息都有一个类别。每个topic可以有多个生产者向他发送消息，也可以有多个消费者去消费消息

partition

每个topic 可以划分多个分区（每个topic至少有一个分区），同一个topic下的不同分区包含的消息是不同的。每个消息在被添加到分区时，都会被分配一个offset，它是消息再此分区中的唯一编号，kafka通过offset保证消息在分区内的顺序，offset 的顺序不跨分区，即kafka只保证在同一个分区内的消息是有序的。

topic 和 partition 存储

partition 是以文件的形式存储在文件系统中，比如创建一个名为 demo 的topic ，其中有三个partition ，那么在kafka 的数据目录中，就有3三目录 ，demo-0~2，命名规则：topicname-partitionid

kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic demo

kafka 消息分发策略

消息是kafka 中最基本的数据单元，在kafka中，一条消息有 key 、value 两部分组成，在发送一条消息时，我们可以指定这个 key，那么 producer 会根据 key 和partition 机制来判断当前这条消息应该发送并存储到哪个partition中；我们可以根据需要进行扩展producer 的partition 机制

/**
 * @author : guaoran
 * @Description : <br/>
 *  自定义消息分区算法
 * @date :2019/1/15 13:35
 */
public class TopicPartitionDemo implements Partitioner {
    private final Random random = new Random();
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitionInfoList = cluster.partitionsForTopic(topic);
        //指定发送的分区值
        int partitionNum = 0;
        if(key==null){
            // 随机分区
            partitionNum = random.nextInt(partitionInfoList.size());
        }else{
            Math.abs((key.hashCode())%partitionInfoList.size());
        }
        System.err.println("topic="+topic+",key="+key+",value="+value+",partitionNum="+partitionNum);
        return partitionNum;
    }
}

消息默认的分发机制

默认情况下，kafka 采用的是hash 取模的分区算法。如果key 为null，则会随机分配一个分区。这个随机是在参数”metadata.max.age.ms” 的时间范围内随机选择一个。对于这个时间段内，如果key为null，则只会发送到唯一的分区。这个值默认情况下是10分钟更新一次。

消费端消费指定的分区

//消费指定分区的时候，不需要再订阅 
//consumer.subscribe(Collections.singletonList(topic));
// todo 只消费分区 0 的消息
TopicPartition partitionDemo = new TopicPartition(topic,0);
consumer.assign(Arrays.asList(partitionDemo));

消息的消费原理

在实际生产过程中，每个topic都会有多个partition，多个partition的好处在于，一方面能够对broker上的数据进行分片有效减少消息的容量从而提升io性能。另一方面，为了提高消费端的消费能力，一般会通过多个consumer 去消费同一个topic，也就是消费端的负载均衡机制。

在group.id相同的consumer进行消费同一个topic时，一个consumer消费过得数据在另一consumer中不会被消费到，那么同一个consumer group 里面的consumer 去消费数据的时候，会根据分片进行分配消费分区的数据。如果有三个partition ，同时启动三个group.id 相同的consumer去同时消费同一个topic，最终的结果是三个consumer 会分别消费一个partition 的数据。

分区分配策略

在kafka中存在两种分区分配策略，一种是Range（默认），一种是RoundRobin（轮询）。通过partition.assignment.strategy 参数来设置。

Range strategy 范围分区

Range 策略是对每个主题而言的，首先对同一个主题里面的分区按照序号进行排序。并对消费者按照字母顺序进行排序。假设有10个分区，3个消费者，排完序的分区将会是0-9;消费者线程排完序是C0-0,C1-1,C2-2 。然后将partitions的个数除于消费者线程的总数来决定每个消费者线程将会消费几个分区。如果除不尽，则前面的消费者会多消费一个分区。所以最终结果是：C0消费 0-3分区，C1消费4-6分区，C2消费7-9分区。

如果同时消费两个主题的话，分区数相同，消费者相同，此时，C0消费者比其他消费者线程多消费2个分区，这就是Range Strategy 的一个弊端。最好是分区数是消费者的整数倍。

RoundRobin strategy 轮询分区

轮询分区策略是把所有的partition 和所有consumer 都列出来，然后按照hashcode进行排序。最后通过轮询算法分配partition给消费者。如果所有consumer实例的订阅都是相同的，那么partition会均匀分布。

使用轮询分区策略必须满足两个条件

1. 每个主题的消费者实例具有相同数量的流
2. 每个消费者订阅的主题必须是相同的。

触发分区分配策略的条件

当出现以下几种情况时，kafka会进行一次分区分配操作，即 kafka consumer 的rebalance

1. 同一个consumer group 内新增了消费者
2. 消费者离开当前的consumer group ，如：主动停机或宕机
3. topic 分区数量发生了变化

谁来执行Rebalance 以及管理 consumer 的group ？

kafka 提供了一个角色：coordinator 来执行对于consumer group的管理 ，当consumer group 的第一个 consumer 启动的时候，它会去跟 kafka server 确定谁是他们组的 coordinator 。之后该group 内所有成员都会和该coordinator 进行协调通信。

确定 coordinator

消费者向kafka 集群中的任意一个broker 发送一个 GroupCoordinatorRequest请求，服务端会返回一个负载最小的broker 节点的id，并将 broker 设置为 coordinator

JoinGroup 的过程

在rebalance 之前，需要保证 coordinator是已经确定好了的，整个Rebalance 的过程分为两个步骤，join 和 sync 。

join：表示加入到consumer group 中，在这一步中，所有成员都会想 coordinator 发送joinGroup的请求。一旦所有成员都发送了joinGroup请求，那么 coordinator会选择一个consumer 担任leader 角色，并把组成员信息和订阅信息发送消费者，并返回分区策略给leader。

Synchronizing GroupGroup StateState 阶段

leader 收到 coordinator 的分区策略后确定分配方案，将消费者对应的partition分配方案同步给consumer group中的所有consumer。

步骤

1. 消费者向任意一个broker 发送消息， broker 返回一个最小的broker id ，作为coordinator
2. 消费者向coordinator 发送 JoinGroup 请求，coordinator 选择 一个消费者作为leader，返回给leader 分区策略
3. 消费者leader 根据分区策略确定分区方案后，向 coordinator 发送 Synchronizing GroupGroup 请求， 就是告知broker 消费者分别消费哪些个分区。