什么是kafka

kafka基于scala和java语言开发，设计中大量使用了批量处理和异步的思想，最高可以每秒处理百万级别的消息，是用于构建实时数据管道和流的应用程序。

kafka的应用场景

kafka是一个分布式流式处理平台。流平台具有三个关键功能：

消息队列：发布和订阅消息流，这个功能类似于消息队列，这也是kafka被归类为消息队列的原因。容错的持久方式存储记录消息流：kafka会把消息持久化到磁盘，有效避免消息丢失的风险。流式处理平台：在消息发布的时候进行处理，kafka提供了一个完整的流式处理类库。

kafka主要有两大应用场景：

消息队列：建立实时流数据管道，可靠地在系统或应用程序之间获取数据。数据处理：构建实时的流数据处理程序来转换或处理数据流。

注： kafka在2.8预览版中，采用raft元数据模式，取消了对zookeeper的依赖。

kafka的版本里程碑

版本号备注0.8引入了副本机制，成为了一个真正意义上完备的分布式高可靠消息队列解决方案0.8.2新版本 producer api，即需要指定 broker 地址的 producer0.9增加了基础的安全认证 / 权限，java 重写了新版本消费者 api0.10引入了 kafka streams0.11提供幂等性 producer api 以及事务（transaction） api，对 kafka 消息格式做了重构。1.0kafka streams 的各种改进2.0kafka streams 的各种改进

kafka的优势

高吞吐、低延时：这是 kafka 显著的特点，kafka 能够达到百万级的消息吞吐量，延迟可达毫秒级。持久化存储：kafka 的消息最终持久化保存在磁盘之上，提供了顺序读写以保证性能，并且通过 kafka 的副本机制提高了数据可靠性。分布式可扩展：kafka的数据是分布式存储在不同broker节点的，以topic组织数据并且按partition进行分布式存储，整体的扩展性都非常好。高容错性：集群中任意一个 broker 节点宕机，kafka 仍能对外提供服务。

kafka基本结构

kafka具有四个核心api：

producer api： 发布消息到1个或多个topic（主题）中。consumer api：来订阅一个或多个topic，并处理产生的消息。streams api：充当一个流处理器，从1个或多个topic消费输入流，并生产一个输出流到1个或多个输出topic，有效地将输入流转换到输出流。connector api：可构建或运行可重用的生产者或消费者，将topic连接到现有的应用程序或数据系统。例如，连接到关系数据库的连接器可以捕获表的每个变更。

kafka的关键术语

producer：消息和数据的生产者，向kafka的一个topic发布消息的进程/代码/服务。consumer：消息和数据的消费者，订阅数据（topic）并且处理发布的消息的进程/代码/服务。consumer group：对于同一个topic，会广播给不同的group。在一个group中，一条消息只能被消费组中一个的consumer消费。

consumer group中不能有比partition数量更多的消费者，否则多出的消费者一直处于空等待，不会收到消息。

topic：每条发布到kafka集群的消息都有一个类别，这个类别被称为topic。作用是对数据进行区分、隔离。broker：kafka集群中的每个kafka节点。保存topic的一个或多个partition。partition：物理概念，kafka下数据储存的基本单元。一个topic数据，会被分散存储到多个partition，每一个partition都是一个顺序的、不可变的消息队列，并且可以持续的添加消息。

注：

每一个topic的信息被切分为多个partitions。若partition数量设置成1个，则可以保证消息消费的顺序性。如果某topic有n个partition，集群有n个broker，那么每个broker存储该topic的一个partition。如果某topic有n个partition，集群有(n+m)个broker，那么其中有n个broker存储该topic的一个partition，剩下的m个broker不存储该topic的partition数据。如果某topic有n个partition，集群中broker数目少于n个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致kafka集群数据不均衡。当broker收到消息，根据分区算法选择将其存储到哪一个 partition。其路由机制为优先按照指定partition来路由；若未指定patition但指定key，则通过对key的value进行hash选出一个patition；如果patition和key都未指定，则轮询选出一个patition。

offt：偏移量，分区中的消息位置，由kafka自身维护，consumer消费时也要保存一份offt以维护消费过的消息位置。replication：同一个partition可能会有多个副本，多个副本之间数据是一样的，增加容错性与可扩展性。

注：

当集群中的有br太祖oker挂掉的情况，系统可以主动的使用replication提供服务。系统默认设置每一个topic的replication系数为1，可以在创建topic时单独设置。replication的基本单位是topic的partition。所有的读和写都由leader进，followers只是做为数据的备份。follower必须能够及时复制leader的数据。

replication leader：一个partition的多个副带一的成语本上，需要一个leader负责该partition上与producer和consumer交互。一个partition只对应一个replication leader。replication follower：follower跟随leader，所有写请求都会广播给所有follower，follower与leader保持数据同步。replicamanager：负责管理当前broker所有分区和副本的信息，处理kafkacontroller发起的一些请求，副本状态的切换、添加/读取消息等。rebalance。消费者组内某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程。rebalance是kafka消费者端实现高可用的重要手段。

kafka通过zookeepe阳光体育活动总结r管理集群配置，选举leader，以及在consumer group发生变化时进行rebalance。

店长的职责

kafka的复制机制

如何将所有replication均匀分布到整个集群

为了更好的做负载均衡，kafka尽量将所有的partition均匀分配到整个集群上。一个典型的部署方式是一个topic的partition数量大于broker的数量。同时为了提高kafka的容错能力，也需要将同一个partition的replication尽量分散到不同的机器。如果所有的replication都在同一个broker上，那一旦该broker宕机，该partition的所有replication都无法工作，也就达不到ha的效果。同时，如果某个broker宕机了，需要保证它上面的负载可以被均匀的分配到其它幸存的所有broker上。

kafka分配replication的算法如下：

将所有broker（假设共n个broker）和待分配的partition排序。将第i个partition分配到第（i % n）个broker上。将第i个partition的第j个replication分配到第（(i + j) % n）个broker上。

hw高水位与leo

hw是high watermark的缩写，俗称高水位，它标识了一个特定的消息偏移量（offt），消费者只能拉取到这个offt之前的消息。

如图所示，它代表一个日志文件，这个日志文件中有 9 条消息，第一条消息的offt（logstartofft）为0，最后一条消息的offt为8，offt为9的消息用虚线框表示，代表下一条待写入的消息。日志文件的hw为6，表示消费者只能拉取到offt在0至5之间的消息，而offt为6的消息对消费者而言是不可见的。

leo是log end offt的缩写，它标识当前日志文件中下一条待写入消息的offt，图中offt为9的位置即为当前日志文件的leo，leo的大小相当于当前日志分区中最后一条消息的offt值加1。分区isr集合中的每个副本都会维护自身的leo，而isr集合中最小的leo即为分区的hw，对消费者而言只能消费hw之前的消息。

isr副本集合

isr全称是“in-sync replicas”，是分区中正在与leader副本进行同步的replication列表。正常情况下isr必定包含leader副本。

isr列表是持久化在zookeeper中的，任何在isr列表中的副本都有资格参与leader选举。

isr列表是动态变化的，副本被包含在isr列表中的条件是由参数replica.lag.time.max.ms控制的，参数含义是副本同步落后于leader的最大时间间隔，默认10s，意思就是如果说某个follower所在的broker因为jvm fullgc之类的问题，卡顿相对leader延时超过10s，就会被从 isr 中排除。kafka之所以这样设计，主要是为了减少消息丢失，只有与leader副本进行实时同步的follower副本才有资格参与leader选举，这里指相对实时。

注：

分区中的所有副本统称为ar（assigned replicas）。isr集合是ar集合中的一个子集。与leader副本同步滞后过多的副本（不包括leader副本）组成osr（out-of-sync replicas）

复制机制

如图所示，假设某个分区的isr集合中有3个副本，即一个leader副本和2个follower副本，此时分区的leo和hw都为3。消息3和消息4从生产者发出之后会被先存入leader副本。

在消息写入leader副本之后，follower副本会发送拉取请求来拉取消息3和消息4以进行消息同步。

在同步过程中，不同的follower副本的同步效率也不尽相同。在某一时刻follower1完全跟上了leader副本而follower2只同步了消息3，如此leader副本的leo为5，follower1的leo为5，follower2的leo为4，那么当前分区的hw取最小值4，此时消费者可以消费到offt为0至3之间的消息。

当所有的副本都成功写入了消息3和消息4，整个分区的hw和leo都变为5，因此消费母系氏族者可以消费到offt为4的消息了。

关于读写分离

kafka并不支持读写分区，生产消费端所有的读写请求都是由replication leader副本处理的，replication follower副本的主要工作就是从leader副本处异步拉取消息，进行消息数据的同步，并不对外提供读写服务。

kafka之所以这样设计，主要是为了保证读写一致性，因为副本同步是一个异步的过程，如果当follower副本还没完全和leader同步时，从follower副本读取数据可能会读不到最新的消息。

kafka的消息发送机制

producer采用push模式将消息发布到broker，每条消息都被append到patition中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障kafka吞吐率）。

producer写入消息序列图如下所示：

流程说明：

producer先从zookeeper的”/brokers/…/state”节点找到该partition的leader。producer将消息发送给该leader。leader将消息写入本地log。followers从leader pull消息，写入本地log后leader发送ack。leader收到所有isr中的replica的ack后，增加hw并向producer发送ack。

broker保存消息

每个patition物理上对应一个文件夹（该文件夹存储该patition的所有消息和索引文件）

无论消息是否被消费，kafka都会保留所有消息。有两种策略可以删除旧数据：

基于时间：log.retention.hours=168基于大小：log.retention.bytes=1073741824

consumer消费消息

kafka集群保持所有的消息，直到它们过期（无论消息是否被消费）。实际上消费者所持有的仅有的元数据就是这个offt（偏移量），也就是说offt由消费者来控制：正常情况当消费者消费消息的时候，偏移量也线性的的增加。但是实际偏移量由消费者控制，消费者可以将偏移量重置为更早的位置，重新读取消息。可以看到这种设计对消费者来说操作自如，一个消费者的操作不会影响其它消费者对此log的处理。