Kafak原理

Apache kafka原理

1 分片与副本机制 :

此处的分片指的是对topic中数据进行分片和建立副本, 一个个topic理解为solrCloud中一个个大的索引库

​ 分片机制：主要解决了单台服务器存储容量有限的问题

​ 当数据量非常大的时候，一个服务器存放不了，就将数据分成两个或者多个部分，存放在多台服务器上。每个服务器上的数据，叫做一个分片

image.png

​ 副本：副本备份机制解决了数据存储的高可用问题

​ 当数据只保存一份的时候，有丢失的风险。为了更好的容错和容灾，将数据拷贝几份，保存到不同的机器上。

image.png

5.2 kafka保证数据不丢失机制

5.2.1 保证生产者端不丢失

1） 消息生产分为同步模式和异步模式

2） 消息确认分为三个状态

​ a) 0：生产者只负责发送数据

​ b) 1：某个partition的leader收到数据给出响应

​ c) -1(all)：某个partition的所有副本都收到数据后给出响应

3） 在同步模式下 :

​ a) 生产者等待10S，如果broker没有给出ack响应，就认为失败。

​ b) 生产者重试3次，如果还没有响应，就报错。

4） 在异步模式下

​ a) 先将数据保存在生产者端的buffer中。Buffer大小是2万条。

​ b) 满足数据阈值或者数量(时间)阈值其中的一个条件就可以发送数据。

​ c) 发送一批数据的大小是500条。

如果broker迟迟不给ack，而buffer又满了。

开发者可以设置是否直接清空buffer中的数据。

5.2.2 broker端消息不丢失

broker端的消息不丢失，其实就是用partition副本机制(高可用)来保证。

Producer ack -1(all). 能够保证所有的副本都同步好了数据。其中一台机器挂了，并不影响数据的完整性。

5.2.3 消费端消息不丢失

offset : 偏移量

​ 记录消费者消费到那个数据上

​ 通过offset commit 来保证数据的不丢失，kafka自己记录了每次消费的offset数值，下次继续消费的时候，会接着上次的offset进行消费。

而offset的信息在kafka0.8版本之前保存在zookeeper中，在0.8版本之后保存到topic中，即使消费者在运行过程中挂掉了，再次启动的时候会找到offset的值，找到之前消费消息的位置，接着消费，由于offset的信息写入的时候并不是每条消息消费完成后都写入的，所以这种情况有可能会造成重复消费，但是不会丢失消息。

如何判断某一个消费者的偏移量放置在50个分组中那个组当中: 通过消费者的groupid

​ (groupid对hashCode()) & 50 得到

kafka会存在重复消费的问题:

5.3 消息存储及查询机制

5.3.1 文件存储机制

image.png

​ segment段中有两个核心的文件一个是log,一个是index。 当log文件等于1G时，新的会写入到下一个segment中。

通过下图中的数据，可以看到一个segment段差不多会存储70万条数据。

5.3.2 文件查询机制 :

需求: 读取 offset=368776 的message消息数据

image.png

5.4 生产者数据分发策略

​ kafka在数据生产的时候，有一个数据分发策略。默认的情况使用DefaultPartitioner.class类。

这个类中就定义数据分发的策略。

image.png

数据分发:  ProducerRecord对象
    参数1:  发往topic的名称   参数2: 表示要发送到那个分片上  参数3: key值  参数4: 数据value
    public ProducerRecord(String topic, Integer partition, K key, V value) {
        this(topic, partition, (Long)null, key, value, (Iterable)null);
    }
    1) 执行要往那个分片上发送数据, 如果指定了分片, 那么就不会使用系统默认的分发策略

1） 如果是用户制定了partition，生产就不会调用DefaultPartitioner.partition()方法

​ 数据分发策略的时候，可以指定数据发往哪个partition。

​ 当ProducerRecord 的构造参数中有partition的时候，就可以发送到对应partition上

/**
 * Creates a record to be sent to a specified topic and partition
 *
 * @param topic The topic the record will be appended to
 * @param partition The partition to which the record should be sent
 * @param key The key that will be included in the record
 * @param value The record contents
 */
public ProducerRecord(String topic, Integer partition, K key, V value) {
    this(topic, partition, null, key, value, null);
}

2） 当用户指定key，使用hash算法。如果key一直不变，同一个key算出来的hash值是个固定值。如果是固定值，这种hash取模就没有意义。

Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions

​ 如果生产者没有指定partition，但是发送消息中有key，就key的hash值。

/**
 * Create a record to be sent to Kafka
 * 
 * @param topic The topic the record will be appended to
 * @param key The key that will be included in the record
 * @param value The record contents
 */
public ProducerRecord(String topic, K key, V value) {
    this(topic, null, null, key, value, null);
}      

3） 当用户既没有指定partition也没有key。

​ 使用轮询的方式发送数据。

/**
 * Create a record with no key
 * 
 * @param topic The topic this record should be sent to
 * @param value The record contents
 */
public ProducerRecord(String topic, V value) {
    this(topic, null, null, null, value, null);
}

5.5 消费者负载均衡机制

​ 一个partition可以被一个组中某一个成员消费

​ 所以如果消费组中有多于partition数量的消费者，那么一定会有消费者无法消费数据。