Kafka核心原理

1、Topic的分片和副本机制

分片作用：

解决单台节点容量有限的问题，节点多，效率提升，吞吐量提升。通过分片，将一个大的容器分解为多个小的容器，分布在不同的节点上，从而实现分布式存储。

分片的数量没有限制，与节点数量没有关系，分片数量不会超过总节点数量的三倍。

副本作用：

提升数据的可靠性，副本越多数据越可靠，但是数据冗余越高。

副本数量有限制，最多和节点的数量相等，但是一般构建1~3个之间。

2、Kafka如何保证数据不丢失

数据传输的三个阶段：

生产者生产数据到broker

broker存储数据

消费者从broker上消费数据

<1> 生产端如何保证数据不丢失：

生产数据到broker之后的响应机制

当生产者生产数据到Broker后，Broker应该给于确认响应（ack）。

ack 确认机制，主要有三种方案，分别为0 1 -1(ALL)

0：生产者只管将数据生产到Borker ，不等待Broker返回的ack 信息

1：生产者将数据生产到Broker，需要等待Broker端Topic的对应分片上的主副本接收到消息后，即为成功发送消息。

-1: 生产者将数据生产到broker，需要等待broker端Topic的对应分片所有副本都接收到消息，即为成功发送

生产中一般根据消息重要情况以及生成和消费速率来选择相应的级别。一般来说，重要程度越高的，安全级别越高，速率越高，优先保证安全性，在此基础上，保持平衡。

<2>Broker 如何保证数据不丢失

Broker可以将每个分片的副本数量设置为多个，提供数据的可靠性同时还需要生产端将ACK设置为-1

<3>消费端如何保证数据不丢失

消费者连接kafka集群，kafka收到请求后，首先会根据group_id 查询上一次消费到了哪个消息偏移量，如果没有找到，默认从当前的位置开始消费数据，之前的消息默认不处理，如果找到了，就从记录的消息偏移量位置继续消费数据即可

消费者消费完数据后，会把对应的消息的偏移量信息重新提交给Broker记录

在提交偏移量的时候有两种提交方式：自动提交偏移量，手动提交偏移量

配置自动提交：

  consumer = KafkaConsumer("test",bootstrap_servers=['localhost:9092'],group_id='g_2',enable_auto_commit=True,auto_commit_interval_ms=1000)

手动提交：

  consumer = KafkaConsumer("test",bootstrap_servers=['localhost:9092'],group_id='g_2',enable_auto_commit=False,auto_commit_interval_ms=1000)consumer.commit()  #同步提交
consumer.commit_async()  #异步提交

3、Kafka中生产者的数据分发策略

分发策略：生产者生产数据到Broker的某一Topic，这个数据最终落入到那个分片的副本，即是分发策略

1、Hash策略 -- 支持

def send(self, topic, value=None, key=None, headers=None, partition=None, timestamp_ms=None):

如果发送数据时指定了topic ,value , key, 即是采用hash 策略

相同key 的hash是一样的，会分发到同一个分区

2、随机分发策略---python客户端支持，Java客户端不支持

发送数据时，如果只传递了Topic和value，即是随机分发

3、轮询策略 -- 2.4版本以上修改为粘性策略，2.4版本以下支持，但是这两种方式Java客户端支持，python不支持

4、指定分区的策略 -- 支持

当发送数据时，如果指定了partition参数，即是采用指定分区策略，分区的编号从0 开始

5、自定义分区策略 -- 支持

from kafka.partitioner import DefaultPartitioner

参考Kafka的默认分发策略方法DefaultPartitioner

    def __call__(cls, key, all_partitions, available):"""Get the partition corresponding to key:param key: partitioning key:param all_partitions: list of all partitions sorted by partition ID:param available: list of available partitions in no particular order:return: one of the values from all_partitions or available"""if key is None:if available:return random.choice(available)return random.choice(all_partitions)idx = murmur2(key)idx &= 0x7fffffffidx %= len(all_partitions)return all_partitions[idx]

自定义实现：

class MyPartitioner(object):def __call__(self, key, all_partitions, available):# 实现分发策略return all_partitions[i]kafkaProducer = KafkaProducer(bootstrap_servers=[],ack=-1,partitioner=MyPartitioner()
)