Kafka简介与基础原理

一：初识kafka

　　对于大型的分布式系统来说，消息中间件可以说是必不可少的，它的核心功能有解耦、异步、削峰， 之前说的RabbitMQ就是其中一种，而kafka则是消息中间件的又一利器。

首先看一下官网（http://kafka.apache.org/intro）的介绍：Apache Kafka is a distributed streaming platform。

流平台具有三个关键功能：

发布和订阅记录流，类似于消息队列或企业消息传递系统。
以容错的持久方式存储记录流。
处理记录流。

Kafka是一款基于发布与订阅的消息系统。

二：kafka核心概念

1：主题和分区日志：

　　kafka中消息订阅和发布都是基于某个主题（topic）的，一个主题中又可以被分为很多分区（partition），kafka无法保证整个主题内消息的顺序但是可以保证分区消息的顺序性。

当消息被写入时，如果键为null，默认的分区器会使用轮询算法均衡的把消息写入到各个分区上，如果键不为null，默认分区器会根据键的散列值映射到特定的分区上，同一个键的消息一定是会被写入到同一个分区上的。因此创建主题时就确认好分区，不要后在主题下新增分区。

来看一下分区的实现：

 1 /**
 2      * computes partition for given record.
 3      * if the record has partition returns the value otherwise
 4      * calls configured partitioner class to compute the partition.
 5      */
 6     private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
 7         Integer partition = record.partition();
 8         return partition != null ?
 9                 partition :
10                 partitioner.partition(
11                         record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
12     }

　第8行的三目运算符表示：如果消息 ProducerRecord 指定了partition字段，那么就不需要分区器。

　　　　　　　　　　　如果消息 ProducerRecord 没有指定partition字段，那么就需要依赖分区器，根据key这个字段来计算partition的值, 分区器的作用就是为消息分配分区。

Kafka默认的分区器是DefaultPartitioner，它实现了Partitioner接口，当然我们也可以去实现Partitioner接口来自定义自己的分区器。

并将 props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "自定义分区器类的全限定名")，放入配置中。

我们看一看DefaultPartitioner的源码实现：

 1 /**
 2      * Compute the partition for the given record.
 3      *
 4      * @param topic The topic name
 5      * @param key The key to partition on (or null if no key)
 6      * @param keyBytes serialized key to partition on (or null if no key)
 7      * @param value The value to partition on or null
 8      * @param valueBytes serialized value to partition on or null
 9      * @param cluster The current cluster metadata
10      */
11     public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
12         List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
13         int numPartitions = partitions.size();
14         if (keyBytes == null) {
15             int nextValue = nextValue(topic);
16             List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
17             if (availablePartitions.size() > 0) {
18                 int part = Utils.toPositive(nextValue) % availablePartitions.size();
19                 return availablePartitions.get(part).partition();
20             } else {
21                 // no partitions are available, give a non-available partition
22                 return Utils.toPositive(nextValue) % numPartitions;
23             }
24         } else {
25             // hash the keyBytes to choose a partition
26             return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
27         }
28     }
29 
30     private int nextValue(String topic) {
31         AtomicInteger counter = topicCounterMap.get(topic);
32         if (null == counter) {
33             counter = new AtomicInteger(ThreadLocalRandom.current().nextInt());
34             AtomicInteger currentCounter = topicCounterMap.putIfAbsent(topic, counter);
35             if (currentCounter != null) {
36                 counter = currentCounter;
37             }
38         }
39         return counter.getAndIncrement();
40     }

View Code

参数中的keyBytes其实就是 key.serializer 将topic和key序列化后的结果，我之前用的StringSerializer 就是等于 key.getBytes(encoding);

keyBytes不为空，则使用称之为murmur的Hash算法（非加密型Hash函数，具备高运算性能及低碰撞率）来计算分区分配。
keyBytes为空，则用轮询的方式选择一个分区。nextValue方法可以理解为是在消息记录中没有指定key的情况下，需要生成一个数用来代替key的hash值方法就是最开始先生成一个随机数，之后在这个随机数的基础上每次请求时均进行+1的操作。

2：生产者

　　生产者一般是将一个消息发布到一个自己选择的主题上，在某些情况下，生产者可以把消息发布到指定的分区，这是通过消息键和分区器实现的。

kafka的生产者发送消息的主要步骤：

2.1：创建一个kafka生产者的属性（http://kafka.apache.org/documentation.html#producerconfigs）

bootstrap.servers：此属性为设置broker的地址，格式为host:port

key.serializer和value.serializer：broker希望接受到的消息的键和值都是字节数组，生产者接口如果想要将java对象类型作为键值发送给broker，

　　　那么生产者需要知道如何将这些java对象转化为字节数组。

acks：指定必须有多少个分区副本收到消息，生产者才认为消息写入成功了。此参数选项有：

　　0：生产者在成功写入消息前，并不需要等待服务器的任何响应，特点是发送速度快，吞吐量高，但可能出现消息丢失。

　　1：需要集群的首领节点收到消息，生产者才会收到一个来自服务器的成功响应。

　　all：所有参与复制的节点都收到消息后，生产者才会收到一个来自服务器的成功响应，特点是最安全，但延迟比acks=1更高。

3：消费者

　　消费者主动从broker读取消息，若干个消费者可以组成一个消费者组（consumer groups），一个消费者组订阅的肯定是同一个主题，

同一个主题下的所有分区消息会被不同的消费者组分别消费（保证不同消费者组都能读取到所有消息），

而同一个消费者组内的消费者不会同时读取一个分区的消息（一个消费者组内不会重复消费同一个消息）。

消费效率最高的情况是partition和consumer groups下的消费者数量相同。这样确保每个consumer单独消费一个partition，

当同组的consumer多于partition时，就会有consumer闲置。

消息的提交

当消费者读取（poll方法）分区的消息后，会更新分区当前位置，而消费者就是基于每个分区最后一次提交的偏移量开始继续消费消息的，所有提交偏移量的方式对消息的处理有很大影响。

1：自动提交

　　enable.auto.commit=true，那么消费者在每个提交间隔（auto.commit.interval.ms）后会主动将接收到的最大偏移量提交上去。这种简便的方式带来的问题是，消费者发生再均衡时会导致上一次提交到再均衡之间的消息被重复处理。

2：手动提交偏移量（commitSync）

　　此方法简单可靠，不用基于时间间隔，而是让程序自己决定提交偏移量，如果发生再均衡，只是最近一次提交到再均衡直接的消息会被重复消费。

3：异步提交（commitAsync）

　　手动提交，程序会阻塞在提交方法上，等待broker对提交请求的返回，异步提交在成功或者遇到无法恢复的错误之前不会一直重试，因为一个更大的偏移量可能已经提交成功，此时重试成功会导致额外的消息重复消费。commitAsync(OffsetCommitCallback callback)也支持回调，可以用来记录提交错误的日志和偏移量，或者根据自身业务决定是否需要重试提交，重试提交一定要注意提交的顺序，可以避免消息被重复消费。

4：同步异步组合提交

 1 try {
 2             while (true) {
 3                 ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
 4                 for (ConsumerRecord record : records) {
 5                     try {
 6                         System.out.println("接受消息：" + record.value());
 7                     } catch (Exception e) {
 8                         e.printStackTrace();
 9                     }
10                     kafkaConsumer.commitAsync();
11                 }
12             }
13 
14         } catch (Exception e) {
15             e.printStackTrace();
16         } finally {
17             try {
18                 kafkaConsumer.commitSync();
19             } finally {
20                 kafkaConsumer.close();
21             }
22         }

第10行的异步提交，如果一切正常，那么commitAsync()提交速度最快

第18行同步提交，如果关闭消费者，那么同步提交会一直重试，直到提交成功或者发生无法恢复的错误。

4：复制

　　复制功能是保证kafka可用性与持久性的关键，kafka的每个broker上都保存着不同主题和分区的副本（Repalica），

副本有两种类型，活跃的称之为leader，其他的是follower。

生产者和消费者的请求都会经过leader，leader也知道哪个follower是跟自己是一致的。

而follower不处理客户端的任何请求，当有消息到达leader时，它尝试复制消息，当leader崩溃时，其中一个follower会被提升为leader。

Kafka通过动态维护一组同步到leader的同步副本（ISR）。只有这组副本才有资格当选leader。直到所有异步副本都收到对Kafka分区的写入后，该写入才被视为已提交。

不完全leader选举：Kafka关于数据丢失的保证是基于至少一个保持同步的副本。如果所有复制分区的节点都死了，则此保证不再成立。此时系统需要作出抉择：

等待ISR中的副本复活，然后选择该副本作为领导者（希望它仍然拥有所有数据）。
选择第一个作为领导者复活的副本（不一定在ISR中，可能是不同步的副本）。

而这两种选择带来的影响是，1导致可用性较低，必须等待原leader恢复可用，2要承担丢失数据和数据不一致的风险。

当写入Kafka时，生产者可以选择是acks为0,1还是all（-1），请注意，acks=all不能保证已分配的副本的全部集合都已收到该消息。

默认情况下，当acks = all时，所有当前的同步副本都收到消息后立即进行确认。例如，如果一个主题仅配置了两个副本，而一个失败（即仅保留一个同步副本），则指定acks = all的写入将成功。

但是，如果其余副本也失败，则这些写操作可能会丢失。尽管这确保了分区的最大可用性，但是对于某些倾向于持久性而不是可用性的用户而言，此行为可能是不希望的。因此，

禁用不完全的leader选举-如果所有副本都变得不可用，则该分区将保持不可用，直到最新的leader再次变得可用为止。与消息丢失的风险相比，这实际上更倾向于不可用性。
指定最小ISR大小-如果ISR的大小大于某个最小值，分区将仅接受写操作，以防止丢失仅写入单个副本的消息，该消息随后将变得不可用。仅当生产者使用acks = all并保证至少有这么多同步副本确认该消息时，此设置才生效。此设置在一致性和可用性之间进行权衡。最小ISR大小的较高设置可确保更好的一致性，因为可以确保将消息写入更多副本，从而降低了丢失消息的可能性。但是，这会降低可用性，因为如果同步副本的数量下降到最小阈值以下，则分区将无法进行写操作。

三：参考文献

1：http://kafka.apache.org/

2：Kafka权威指南

3：Apache Kafka核心概念