kafka(一)--概念理解

约定不等于承诺〃 2022-10-09 04:51 253阅读 0赞

**目录**

第一篇文章部分信息

什么是卡夫卡？

卡夫卡术语

Kafka 主题剖析

分区和代理

生产者

消费者和消费者团体

第二篇文章部分信息

2.Kafka文件存储机制

--------------------

# 第一篇文章部分信息 #

原文：[https://sookocheff.com/post/kafka/kafka-in-a-nutshell/][https_sookocheff.com_post_kafka_kafka-in-a-nutshell]

Kafka 是一个消息系统。就是这样。那么为什么所有的炒作？实际上，消息传递是用于在系统之间移动数据的非常重要的基础设施。要了解原因，让我们看看没有消息传递系统的数据管道。

该系统从 Hadoop 开始进行存储和数据处理。Hadoop 在没有数据的情况下不是很有用，因此使用 Hadoop 的第一阶段是获取数据。

![7f50ac0f301e32cb70c0d75cbacfb1ae.png][]

**将数据引入 Hadoop**

到目前为止，没什么大不了的。不幸的是，在现实世界中，数据并行存在于许多系统上，所有这些系统都需要与 Hadoop 进行交互并相互交互。情况很快变得更加复杂，最终形成一个系统，其中多个数据系统通过多个通道相互通信。这些渠道中的每一个都需要自己的自定义协议和通信方法，并且在这些系统之间移动数据成为开发团队的全职工作。

![3b49a49c726d53a1d6d90498f75ecd27.png][]

**在系统之间移动数据**

我们再来看这张图，使用 Kafka 作为中央消息总线。所有传入的数据首先放在 Kafka 中，所有传出的数据都从 Kafka 中读取。Kafka 集中了数据生产者和数据消费者之间的通信。

![f721d7bd05789f4aabebd2a5b69330e7.png][]

**在系统之间移动数据**

## **什么是卡夫卡？** ##

Kafka 是一个分布式消息传递系统，通过发布-订阅模型提供快速、高度可扩展和冗余的消息传递。Kafka 的分布式设计赋予了它几个优势。首先，Kafka 允许大量永久或临时消费者。其次，Kafka 具有高可用性和对节点故障的弹性，并支持自动恢复。在现实世界的数据系统中，这些特性使 Kafka 非常适合大规模数据系统组件之间的通信和集成。

## **卡夫卡术语** ##

Kafka 的基本架构是围绕几个关键术语组织的：主题、生产者、消费者和代理。

所有 Kafka 消息都组织成*主题*。如果您希望发送消息，请将其发送到特定主题；如果您希望阅读消息，则从特定主题阅读。一个*消费*拉动的消息掀起了卡夫卡的话题，而*生产者*将消息传递到卡夫卡的话题。最后，Kafka 作为分布式系统，运行在集群中。集群中的每个节点都称为 Kafka *broker*。

## **Kafka 主题剖析** ##

Kafka 主题分为多个*分区*。分区允许您通过将特定主题中的数据拆分到多个代理来并行化主题——每个分区可以放置在单独的机器上，以允许多个消费者并行读取一个主题。消费者也可以并行化，以便多个消费者可以从主题中的多个分区读取，从而实现非常高的消息处理吞吐量。

分区内的每条消息都有一个标识符，称为其*偏移量*。作为不可变序列的消息排序的偏移量。Kafka 为您维护此消息排序。消费者可以从特定偏移量开始读取消息，并允许从他们选择的任何偏移点读取消息，允许消费者在他们认为合适的任何时间点加入集群。鉴于这些限制，Kafka 集群中的每条特定消息都可以由一个元组唯一标识，该元组由消息的主题、分区和分区内的偏移量组成。

![log-anatomy.png][]

**日志解剖**

查看分区的另一种方法是作为日志。数据源将消息写入日志，一个或多个消费者在他们选择的时间点从日志中读取。在下图中，数据源正在写入日志，而消费者 A 和 B 正在以不同的偏移量从日志中读取。

![da16f907e8ba5f8560bdee21b3873e51.png][]

**数据记录**

Kafka 将消息保留一段可配置的时间，消费者可以相应地调整他们的行为。例如，如果 Kafka 配置为将消息保留一天，而消费者停机时间超过一天，则消费者将丢失消息。但是，如果消费者停机一个小时，它可以从最后一个已知的偏移量开始再次读取消息。从 Kafka 的角度来看，它没有关于消费者从主题中读取的内容的状态。

## **分区和代理** ##

每个 Broker 拥有多个分区，这些分区中的每一个都可以是一个主题的领导者或副本。对主题的所有写入和读取都经过领导者，领导者协调使用新数据更新副本。如果领导失败，副本将接替新领导。

![分区.png][.png]

**分区和代理**

## **生产者** ##

生产者写入单个领导者，这提供了一种负载平衡生产的方法，以便每个写入都可以由单独的代理和机器提供服务。在第一个图像中，生产者正在写入主题的分区 0，分区 0 复制写入可用副本。

![生产到分区.png][.png 1]

**生产者写入分区。**

在第二个图像中，生产者正在写入主题的分区 1，分区 1 复制写入可用副本。

![生产到第二个分区.png][.png 2]

**生产者写入第二个分区。**

由于每台机器负责每次写入，因此整个系统的吞吐量增加。

## **消费者和消费者团体** ##

消费者从任何单个分区读取，允许您以类似于消息生产的方式扩展消息消费的吞吐量。对于给定的主题，消费者也可以被组织成消费者组——组中的每个消费者都从一个唯一的分区中读取数据，并且该组作为一个整体消费来自整个主题的所有消息。如果消费者的数量多于分区，那么一些消费者将空闲，因为他们没有可供读取的分区。如果您的分区比消费者多，那么消费者将收到来自多个分区的消息。如果您有相同数量的消费者和分区，则每个消费者都会从一个分区中按顺序读取消息。

从下面的图片[卡夫卡文档][Link 1]描述了一个主题的多个分区的情况。服务器 1 包含分区 0 和 3，服务器 2 包含分区 1 和 2。我们有两个消费者组，A 和 B。A 由两个消费者组成，B 由四个消费者组成。消费者组 A 有四个分区的两个消费者——每个消费者从两个分区读取。另一方面，消费者组 B 具有与分区相同数量的消费者，并且每个消费者都从一个分区读取数据。

![消费者组.png][.png 3]

**消费者和消费者团体**

# 第二篇文章部分信息 #

来自2：[https://www.cnblogs.com/dreamroute/p/13092117.html][https_www.cnblogs.com_dreamroute_p_13092117.html]

**Kafka**

Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。

**1.前言**

消息队列的性能好坏，其[文件存储][Link 2]机制设计是衡量一个消息队列服务技术水平和最关键指标之一。下面将从Kafka文件存储机制和物理结构角度，分析Kafka是如何实现高效文件存储，及实际应用效果。

**1.1 Kafka的特性:**

\- 高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作。

\- 可扩展性：kafka集群支持热扩展

\- 持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失

\- 容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）

\- 高并发：支持数千个客户端同时读写

**1.2 Kafka的使用场景：**

\- 日志收集：一个公司可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、[Hbase][]、Solr等。

\- 消息系统：解耦和生产者和消费者、缓存消息等。

\- 用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。

\- 运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

\- 流式处理：比如spark streaming和storm

\- 事件源

**1.3 Kakfa的设计思想**

\- **Kakfa Broker Leader的选举：**Kakfa Broker集群受Zookeeper管理。所有的Kafka Broker节点一起去Zookeeper上注册一个临时节点，因为只有一个Kafka Broker会注册成功，其他的都会失败，所以这个成功在Zookeeper上注册临时节点的这个Kafka Broker会成为Kafka Broker Controller，其他的Kafka broker叫Kafka Broker follower。（这个过程叫Controller在ZooKeeper注册Watch）。这个Controller会监听其他的Kafka Broker的所有信息，如果这个kafka broker controller宕机了，在zookeeper上面的那个临时节点就会消失，此时所有的kafka broker又会一起去Zookeeper上注册一个临时节点，因为只有一个Kafka Broker会注册成功，其他的都会失败，所以这个成功在Zookeeper上注册临时节点的这个Kafka Broker会成为Kafka Broker Controller，其他的Kafka broker叫Kafka Broker follower。例如：一旦有一个broker宕机了，这个kafka broker controller会读取该宕机broker上所有的partition在zookeeper上的状态，并选取ISR列表中的一个replica作为partition leader（如果ISR列表中的replica全挂，选一个幸存的replica作为leader; 如果该partition的所有的replica都宕机了，则将新的leader设置为-1，等待恢复，等待ISR中的任一个Replica“活”过来，并且选它作为Leader；或选择第一个“活”过来的Replica（不一定是ISR中的）作为Leader），这个broker宕机的事情，kafka controller也会通知zookeeper，zookeeper就会通知其他的kafka broker。

**这里曾经发生过一个bug，TalkingData使用Kafka0.8.1的时候，kafka controller在Zookeeper上注册成功后，它和Zookeeper通信的timeout时间是6s，也就是如果kafka controller如果有6s中没有和Zookeeper做心跳，那么Zookeeper就认为这个kafka controller已经死了，就会在Zookeeper上把这个临时节点删掉，那么其他Kafka就会认为controller已经没了，就会再次抢着注册临时节点，注册成功的那个kafka broker成为controller，然后，之前的那个kafka controller就需要各种shut down去关闭各种节点和事件的监听。但是当kafka的读写流量都非常巨大的时候，TalkingData的一个bug是，由于网络等原因，kafka controller和Zookeeper有6s中没有通信，于是重新选举出了一个新的kafka controller，但是原来的controller在shut down的时候总是不成功，这个时候producer进来的message由于Kafka集群中存在两个kafka controller而无法落地。导致数据淤积。**

**这里曾经还有一个bug，TalkingData使用Kafka0.8.1的时候，当ack=0的时候，表示producer发送出去message，只要对应的kafka broker topic partition leader接收到的这条message，producer就返回成功，不管partition leader 是否真的成功把message真正存到kafka。当ack=1的时候，表示producer发送出去message，同步的把message存到对应topic的partition的leader上，然后producer就返回成功，partition leader异步的把message同步到其他partition replica上。当ack=all或-1，表示producer发送出去message，同步的把message存到对应topic的partition的leader和对应的replica上之后，才返回成功。但是如果某个kafka controller** **切换的时候，会导致partition leader的切换（老的** **kafka controller上面的partition leader会选举到其他的kafka broker上）,但是这样就会导致丢数据。**

\- **Consumergroup：**各个consumer（consumer 线程）可以组成一个组（Consumer group ），partition中的每个message只能被组（Consumer group ）中的一个consumer（consumer 线程）消费，如果一个message可以被多个consumer（consumer 线程）消费的话，那么这些consumer必须在不同的组。Kafka不支持一个partition中的message由两个或两个以上的同一个consumer group下的consumer thread来处理，除非再启动一个新的consumer group。所以如果想同时对一个topic做消费的话，启动多个consumer group就可以了，但是要注意的是，这里的多个consumer的消费都必须是顺序读取partition里面的message，新启动的consumer默认从partition队列最头端最新的地方开始阻塞的读message。它不能像AMQ那样可以多个BET作为consumer去互斥的（for update悲观锁）并发处理message，这是因为多个BET去消费一个Queue中的数据的时候，由于要保证不能多个线程拿同一条message，所以就需要行级别悲观所（for update）,这就导致了consume的性能下降，吞吐量不够。而kafka为了保证吞吐量，只允许同一个consumer group下的一个consumer线程去访问一个partition。如果觉得效率不高的时候，可以加partition的数量来横向扩展，那么再加新的consumer thread去消费。如果想多个不同的业务都需要这个topic的数据，起多个consumer group就好了，大家都是顺序的读取message，offsite的值互不影响。这样没有锁竞争，充分发挥了横向的扩展性，吞吐量极高。这也就形成了分布式消费的概念。

当启动一个consumer group去消费一个topic的时候，无论topic里面有多个少个partition，无论我们consumer group里面配置了多少个consumer thread，这个consumer group下面的所有consumer thread一定会消费全部的partition；即便这个consumer group下只有一个consumer thread，那么这个consumer thread也会去消费所有的partition。因此，最优的设计就是，consumer group下的consumer thread的数量等于partition数量，这样效率是最高的。

同一partition的一条message只能被同一个Consumer Group内的一个Consumer消费。不能够一个consumer group的多个consumer同时消费一个partition。

一个consumer group下，无论有多少个consumer，这个consumer group一定会去把这个topic下所有的partition都消费了。当consumer group里面的consumer数量小于这个topic下的partition数量的时候，如下图groupA,groupB，就会出现一个conusmer thread消费多个partition的情况，总之是这个topic下的partition都会被消费。如果consumer group里面的consumer数量等于这个topic下的partition数量的时候，如下图groupC，此时效率是最高的，每个partition都有一个consumer thread去消费。当consumer group里面的consumer数量大于这个topic下的partition数量的时候，如下图GroupD，就会有一个consumer thread空闲。因此，我们在设定consumer group的时候，只需要指明里面有几个consumer数量即可，无需指定对应的消费partition序号，consumer会自动进行rebalance。

多个Consumer Group下的consumer可以消费同一条message，但是这种消费也是以o（1）的方式顺序的读取message去消费,，所以一定会重复消费这批message的，不能向AMQ那样多个BET作为consumer消费（对message加锁，消费的时候不能重复消费message）

\- **Consumer Rebalance的触发条件：**（1）Consumer增加或删除会触发 Consumer Group的Rebalance（2）Broker的增加或者减少都会触发 Consumer Rebalance

\- **Consumer：** Consumer处理partition里面的message的时候是o（1）顺序读取的。所以必须维护着上一次读到哪里的offsite信息。high level API,offset存于Zookeeper中，low level API的offset由自己维护。一般来说都是使用high level api的。Consumer的delivery gurarantee，默认是读完message先commmit再处理message，autocommit默认是true，这时候先commit就会更新offsite+1，一旦处理失败，offsite已经+1，这个时候就会丢message；也可以配置成读完消息处理再commit，这种情况下consumer端的响应就会比较慢的，需要等处理完才行。

一般情况下，一定是一个consumer group处理一个topic的message。Best Practice是这个consumer group里面consumer的数量等于topic里面partition的数量，这样效率是最高的，一个consumer thread处理一个partition。如果这个consumer group里面consumer的数量小于topic里面partition的数量，就会有consumer thread同时处理多个partition（这个是kafka自动的机制，我们不用指定），但是总之这个topic里面的所有partition都会被处理到的。。如果这个consumer group里面consumer的数量大于topic里面partition的数量，多出的consumer thread就会闲着啥也不干，剩下的是一个consumer thread处理一个partition，这就造成了资源的浪费，因为一个partition不可能被两个consumer thread去处理。所以我们线上的分布式多个service服务，每个service里面的kafka consumer数量都小于对应的topic的partition数量，但是所有服务的consumer数量只和等于partition的数量，这是因为分布式service服务的所有consumer都来自一个consumer group，如果来自不同的consumer group就会处理重复的message了（同一个consumer group下的consumer不能处理同一个partition，不同的consumer group可以处理同一个topic，那么都是顺序处理message，一定会处理重复的。一般这种情况都是两个不同的业务逻辑，才会启动两个consumer group来处理一个topic）。

**如果producer的流量增大，当前的topic的parition数量=consumer数量，这时候的应对方式就是很想扩展：增加topic下的partition，同时增加这个consumer group下的consumer。**

\- **Delivery Mode :** Kafka producer 发送message不用维护message的offsite信息，因为这个时候，offsite就相当于一个自增id，producer就尽管发送message就好了。而且Kafka与AMQ不同，AMQ大都用在处理业务逻辑上，而Kafka大都是日志，所以Kafka的producer一般都是大批量的batch发送message，向这个topic一次性发送一大批message，load balance到一个partition上，一起插进去，offsite作为自增id自己增加就好。但是Consumer端是需要维护这个partition当前消费到哪个message的offsite信息的，这个offsite信息，high level api是维护在Zookeeper上，low level api是自己的程序维护。（Kafka管理界面上只能显示high level api的consumer部分，因为low level api的partition offsite信息是程序自己维护，kafka是不知道的，无法在管理界面上展示 ）当使用high level api的时候，先拿message处理，再定时自动commit offsite+1（也可以改成手动）, 并且kakfa处理message是没有锁操作的。因此如果处理message失败，此时还没有commit offsite+1，当consumer thread重启后会重复消费这个message。但是作为高吞吐量高并发的实时处理系统，at least once的情况下，至少一次会被处理到，是可以容忍的。如果无法容忍，就得使用low level api来自己程序维护这个offsite信息，那么想什么时候commit offsite+1就自己搞定了。

\- **Topic & Partition：**Topic相当于传统消息系统MQ中的一个队列queue，producer端发送的message必须指定是发送到哪个topic，但是不需要指定topic下的哪个partition，因为kafka会把收到的message进行load balance，均匀的分布在这个topic下的不同的partition上（ hash(message) % \[broker数量\] ）。物理上存储上，这个topic会分成一个或多个partition，每个partiton相当于是一个子queue。在物理结构上，每个partition对应一个物理的目录（文件夹），文件夹命名是\[topicname\]\_\[partition\]\_\[序号\]，一个topic可以有无数多的partition，根据业务需求和数据量来设置。在kafka配置文件中可随时更高num.partitions参数来配置更改topic的partition数量，在创建Topic时通过参数指定parittion数量。Topic创建之后通过Kafka提供的工具也可以修改partiton数量。

一般来说，（1）一个Topic的Partition数量大于等于Broker的数量，可以提高吞吐率。（2）同一个Partition的Replica尽量分散到不同的机器，高可用。

**当add a new partition的时候**，partition里面的message不会重新进行分配，原来的partition里面的message数据不会变，新加的这个partition刚开始是空的，随后进入这个topic的message就会重新参与所有partition的load balance

\- **Partition Replica：**每个partition可以在其他的kafka broker节点上存副本，以便某个kafka broker节点宕机不会影响这个kafka集群。存replica副本的方式是按照kafka broker的顺序存。例如有5个kafka broker节点，某个topic有3个partition，每个partition存2个副本，那么partition1存broker1,broker2，partition2存broker2,broker3。。。以此类推**（replica副本数目不能大于kafka broker节点的数目，否则报错。这里的replica数其实就是partition的副本总数，其中包括一个leader，其他的就是copy副本）**。这样如果某个broker宕机，其实整个kafka内数据依然是完整的。但是，replica副本数越高，系统虽然越稳定，但是回来带资源和性能上的下降；replica副本少的话，也会造成系统丢数据的风险。

（1）怎样传送消息：producer先把message发送到partition leader，再由leader发送给其他partition follower。（如果让producer发送给每个replica那就太慢了）

（2）在向Producer发送ACK前需要保证有多少个Replica已经收到该消息：根据ack配的个数而定

（3）怎样处理某个Replica不工作的情况：如果这个部工作的partition replica不在ack列表中，就是producer在发送消息到partition leader上，partition leader向partition follower发送message没有响应而已，这个不会影响整个系统，也不会有什么问题。如果这个不工作的partition replica在ack列表中的话，producer发送的message的时候会等待这个不工作的partition replca写message成功，但是会等到time out，然后返回失败因为某个ack列表中的partition replica没有响应，此时kafka会自动的把这个部工作的partition replica从ack列表中移除，以后的producer发送message的时候就不会有这个ack列表下的这个部工作的partition replica了。

（4）怎样处理Failed Replica恢复回来的情况：如果这个partition replica之前不在ack列表中，那么启动后重新受Zookeeper管理即可，之后producer发送message的时候，partition leader会继续发送message到这个partition follower上。如果这个partition replica之前在ack列表中，此时重启后，需要把这个partition replica再手动加到ack列表中。（ack列表是手动添加的，出现某个部工作的partition replica的时候自动从ack列表中移除的）

\- **Partition leader与follower：**partition也有leader和follower之分。leader是主partition，producer写kafka的时候先写partition leader，再由partition leader push给其他的partition follower。partition leader与follower的信息受Zookeeper控制，一旦partition leader所在的broker节点宕机，zookeeper会冲其他的broker的partition follower上选择follower变为parition leader。

\- **Topic分配partition和partition replica的算法：**（1）将Broker（size=n）和待分配的Partition排序。（2）将第i个Partition分配到第（i%n）个Broker上。（3）将第i个Partition的第j个Replica分配到第（(i + j) % n）个Broker上

**- 消息投递可靠性**

一个消息如何算投递成功，Kafka提供了三种模式：

\- 第一种是啥都不管，发送出去就当作成功，这种情况当然不能保证消息成功投递到broker；

\- 第二种是Master-Slave模型，只有当Master和所有Slave都接收到消息时，才算投递成功，这种模型提供了最高的投递可靠性，但是损伤了性能；

\- 第三种模型，即只要Master确认收到消息就算投递成功；实际使用时，根据应用特性选择，绝大多数情况下都会中和可靠性和性能选择第三种模型

消息在broker上的可靠性，因为消息会持久化到磁盘上，所以如果正常stop一个broker，其上的数据不会丢失；但是如果不正常stop，可能会使存在页面缓存来不及写入磁盘的消息丢失，这可以通过配置flush页面缓存的周期、阈值缓解，但是同样会频繁的写磁盘会影响性能，又是一个选择题，根据实际情况配置。

消息消费的可靠性，Kafka提供的是“At least once”模型，因为消息的读取进度由offset提供，offset可以由消费者自己维护也可以维护在zookeeper里，但是当消息消费后consumer挂掉，offset没有即时写回，就有可能发生重复读的情况，这种情况同样可以通过调整commit offset周期、阈值缓解，甚至消费者自己把消费和commit offset做成一个事务解决，但是如果你的应用不在乎重复消费，那就干脆不要解决，以换取最大的性能。

\- **Partition ack：**当ack=1，表示producer写partition leader成功后，broker就返回成功，无论其他的partition follower是否写成功。当ack=2，表示producer写partition leader和其他一个follower成功的时候，broker就返回成功，无论其他的partition follower是否写成功。当ack=-1\[parition的数量\]的时候，表示只有producer全部写成功的时候，才算成功，kafka broker才返回成功信息。**这里需要注意的是，如果ack=1的时候，一旦有个broker宕机导致partition的follower和leader切换，会导致丢数据。**

\- **message状态**：在Kafka中，消息的状态被保存在consumer中，broker不会关心哪个消息被消费了被谁消费了，只记录一个offset值（指向partition中下一个要被消费的消息位置），这就意味着如果consumer处理不好的话，broker上的一个消息可能会被消费多次。

\- **message持久化**：Kafka中会把消息持久化到本地文件系统中，并且保持o(1)极高的效率。我们众所周知IO读取是非常耗资源的性能也是最慢的，这就是为了数据库的瓶颈经常在IO上，需要换SSD硬盘的原因。但是Kafka作为吞吐量极高的MQ，却可以非常高效的message持久化到文件。这是因为Kafka是顺序写入o（1）的时间复杂度，速度非常快。也是高吞吐量的原因。由于message的写入持久化是顺序写入的，因此message在被消费的时候也是按顺序被消费的，保证partition的message是顺序消费的。一般的机器,单机每秒100k条数据。

\- **message有效期**：Kafka会长久保留其中的消息，以便consumer可以多次消费，当然其中很多细节是可配置的。

\- **Produer :** Producer向Topic发送message，不需要指定partition，直接发送就好了。kafka通过partition ack来控制是否发送成功并把信息返回给producer，producer可以有任意多的thread，这些kafka服务器端是不care的。Producer端的delivery guarantee默认是At least once的。也可以设置Producer异步发送实现At most once。Producer可以用主键幂等性实现Exactly once

\- **Kafka高吞吐量**：Kafka的高吞吐量体现在读写上，分布式并发的读和写都非常快，写的性能体现在以o(1)的时间复杂度进行顺序写入。读的性能体现在以o(1)的时间复杂度进行顺序读取， 对topic进行partition分区，consume group中的consume线程可以以很高能性能进行顺序读。

\- Kafka delivery guarantee(message传送保证)：（1）At most once消息可能会丢，绝对不会重复传输；（2）At least once 消息绝对不会丢，但是可能会重复传输；（3）Exactly once每条信息肯定会被传输一次且仅传输一次，这是用户想要的。

\- **批量发送**：Kafka支持以消息集合为单位进行批量发送，以提高push效率。

\- **push-and-pull** : Kafka中的Producer和consumer采用的是push-and-pull模式，即Producer只管向broker push消息，consumer只管从broker pull消息，两者对消息的生产和消费是异步的。

\- **Kafka集群中broker之间的关系**：不是主从关系，各个broker在集群中地位一样，我们可以随意的增加或删除任何一个broker节点。

\- [**负载均衡**][Link 3]**方面**：Kafka提供了一个 metadata API来管理broker之间的负载（对Kafka0.8.x而言，对于0.7.x主要靠zookeeper来实现负载均衡）。

\- **同步异步**：Producer采用异步push方式，极大提高Kafka系统的吞吐率（可以通过参数控制是采用同步还是异步方式）。

\- **分区机制partition**：Kafka的broker端支持消息分区partition，Producer可以决定把消息发到哪个partition，在一个partition 中message的顺序就是Producer发送消息的顺序，一个topic中可以有多个partition，具体partition的数量是可配置的。partition的概念使得kafka作为MQ可以横向扩展，吞吐量巨大。partition可以设置replica副本，replica副本存在不同的kafka broker节点上，第一个partition是leader,其他的是follower，message先写到partition leader上，再由partition leader push到parition follower上。所以说kafka可以水平扩展，也就是扩展partition。

\- **离线数据装载**：Kafka由于对可拓展的数据持久化的支持，它也非常适合向Hadoop或者数据仓库中进行数据装载。

\- **实时数据与离线数据：**kafka既支持离线数据也支持实时数据，因为kafka的message持久化到文件，并可以设置有效期，因此可以把kafka作为一个高效的存储来使用，可以作为离线数据供后面的分析。当然作为分布式实时消息系统，大多数情况下还是用于实时的数据处理的，但是当cosumer消费能力下降的时候可以通过message的持久化在淤积数据在kafka。

\- **插件支持**：现在不少活跃的社区已经开发出不少插件来拓展Kafka的功能，如用来配合Storm、Hadoop、flume相关的插件。

\- **解耦**: 相当于一个MQ，使得Producer和Consumer之间异步的操作，系统之间解耦

\- **冗余**: replica有多个副本，保证一个broker node宕机后不会影响整个服务

\- **扩展性**: broker节点可以水平扩展，partition也可以水平增加，partition replica也可以水平增加

\- **峰值**: 在访问量剧增的情况下，kafka水平扩展, 应用仍然需要继续发挥作用

\- **可恢复性**: 系统的一部分组件失效时，由于有partition的replica副本，不会影响到整个系统。

\- **顺序保证性**：由于kafka的producer的写message与consumer去读message都是顺序的读写，保证了高效的性能。

\- **缓冲**：由于producer那面可能业务很简单，而后端consumer业务会很复杂并有数据库的操作，因此肯定是producer会比consumer处理速度快，如果没有kafka，producer直接调用consumer，那么就会造成整个系统的处理速度慢，加一层kafka作为MQ，可以起到缓冲的作用。

\- **异步通信**：作为MQ，Producer与Consumer异步通信

## **2.Kafka文件存储机制** ##

**2.1 Kafka部分名词解释如下：**

Kafka 的发布订阅的对象是topic。我们可以为每类数据创建一个topic，把向topic发布消息的客户端称作producer，从topic订阅消息的客户端称作consumer。Producers和consumers可以同时从多个topic读写数据。一个kafka集群由一个或多个broker服务器组成，它负责持久化和备份具体的kafka消息。

*  **Broker**：Kafka节点，一个Kafka节点就是一个broker，多个broker可以组成一个Kafka集群。
 *  **Topic**：一类消息，消息存放的目录即主题，例如page view日志、click日志等都可以以topic的形式存在，Kafka集群能够同时负责多个topic的分发。
 *  **Partition**：topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列
 *  **Segment**：partition物理上由多个segment组成，每个Segment存着message信息
 *  **Producer** : 生产message发送到topic
 *  **Consumer** : 订阅topic消费message, consumer作为一个线程来消费
 *  **Consumer Group**：一个Consumer Group包含多个consumer, 这个是预先在配置文件中配置好的。各个consumer（consumer 线程）可以组成一个组（Consumer group ），partition中的每个message只能被组（Consumer group ） 中的一个consumer（consumer 线程 ）消费，如果一个message可以被多个consumer（consumer 线程 ） 消费的话，那么这些consumer必须在不同的组。Kafka不支持一个partition中的message由两个或两个以上的consumer thread来处理，即便是来自不同的consumer group的也不行。它不能像AMQ那样可以多个BET作为consumer去处理message，这是因为多个BET去消费一个Queue中的数据的时候，由于要保证不能多个线程拿同一条message，所以就需要行级别悲观所（for update）,这就导致了consume的性能下降，吞吐量不够。而kafka为了保证吞吐量，只允许一个consumer线程去访问一个partition。如果觉得效率不高的时候，可以加partition的数量来横向扩展，那么再加新的consumer thread去消费。这样没有锁竞争，充分发挥了横向的扩展性，吞吐量极高。这也就形成了分布式消费的概念。

**2.2 kafka一些原理概念**

**1.持久化**

kafka使用文件存储消息(append only log),这就直接决定kafka在性能上严重依赖文件系统的本身特性.且无论任何OS下,对文件系统本身的优化是非常艰难的.文件缓存/直接内存映射等是常用的手段.因为kafka是对日志文件进行append操作,因此磁盘检索的开支是较小的;同时为了减少磁盘写入的次数,broker会将消息暂时buffer起来,当消息的个数(或尺寸)达到一定阀值时,再flush到磁盘,这样减少了磁盘IO调用的次数.对于kafka而言,较高性能的磁盘,将会带来更加直接的性能提升.

**2.性能**

除磁盘IO之外,我们还需要考虑网络IO,这直接关系到kafka的吞吐量问题.kafka并没有提供太多高超的技巧;对于producer端,可以将消息buffer起来,当消息的条数达到一定阀值时,批量发送给broker;对于consumer端也是一样,批量fetch多条消息.不过消息量的大小可以通过配置文件来指定.对于kafka broker端,似乎有个sendfile系统调用可以潜在的提升网络IO的性能:将文件的数据映射到系统内存中,socket直接读取相应的内存区域即可,而无需进程再次copy和交换(这里涉及到"磁盘IO数据"/"内核内存"/"进程内存"/"网络缓冲区",多者之间的数据copy).

其实对于producer/consumer/broker三者而言,CPU的开支应该都不大,因此启用消息压缩机制是一个良好的策略;压缩需要消耗少量的CPU资源,不过对于kafka而言,网络IO更应该需要考虑.可以将任何在网络上传输的消息都经过压缩.kafka支持gzip/snappy等多种压缩方式.

**3.负载均衡**

kafka集群中的任何一个broker,都可以向producer提供metadata信息,这些metadata中包含"集群中存活的servers列表"/"partitions leader列表"等信息(请参看zookeeper中的节点信息). 当producer获取到metadata信息之后, producer将会和Topic下所有partition leader保持socket连接;消息由producer直接通过socket发送到broker,中间不会经过任何"路由层".

异步发送，将多条消息暂且在客户端buffer起来,并将他们批量发送到broker;小数据IO太多,会拖慢整体的网络延迟,批量延迟发送事实上提升了网络效率;不过这也有一定的隐患,比如当producer失效时,那些尚未发送的消息将会丢失。

**4.Topic模型**

其他JMS实现,消息消费的位置是有prodiver保留,以便避免重复发送消息或者将没有消费成功的消息重发等,同时还要控制消息的状态.这就要求JMS broker需要太多额外的工作.在kafka中,partition中的消息只有一个consumer在消费,且不存在消息状态的控制,也没有复杂的消息确认机制,可见kafka broker端是相当轻量级的.当消息被consumer接收之后,consumer可以在本地保存最后消息的offset,并间歇性的向zookeeper注册offset.由此可见,consumer客户端也很轻量级。

kafka中consumer负责维护消息的消费记录,而broker则不关心这些,这种设计不仅提高了consumer端的灵活性,也适度的减轻了broker端设计的复杂度;这是和众多JMS prodiver的区别.此外,kafka中消息ACK的设计也和JMS有很大不同,kafka中的消息是批量(通常以消息的条数或者chunk的尺寸为单位)发送给consumer,当消息消费成功后,向zookeeper提交消息的offset,而不会向broker交付ACK.或许你已经意识到,这种"宽松"的设计,将会有"丢失"消息/"消息重发"的危险.

**5.消息传输一致**

Kafka提供3种消息传输一致性语义：最多1次，最少1次，恰好1次。

最少1次：可能会重传数据，有可能出现数据被重复处理的情况;

最多1次：可能会出现数据丢失情况;

恰好1次：并不是指真正只传输1次，只不过有一个机制。确保不会出现“数据被重复处理”和“数据丢失”的情况。

at most once: 消费者fetch消息,然后保存offset,然后处理消息;当client保存offset之后,但是在消息处理过程中consumer进程失效(crash),导致部分消息未能继续处理.那么此后可能其他consumer会接管,但是因为offset已经提前保存,那么新的consumer将不能fetch到offset之前的消息(尽管它们尚没有被处理),这就是"at most once".

at least once: 消费者fetch消息,然后处理消息,然后保存offset.如果消息处理成功之后,但是在保存offset阶段zookeeper异常或者consumer失效,导致保存offset操作未能执行成功,这就导致接下来再次fetch时可能获得上次已经处理过的消息,这就是"at least once".

"Kafka Cluster"到消费者的场景中可以采取以下方案来得到“恰好1次”的一致性语义：

最少1次＋消费者的输出中额外增加已处理消息最大编号：由于已处理消息最大编号的存在，不会出现重复处理消息的情况。

**6.副本**

kafka中,replication策略是基于partition,而不是topic;kafka将每个partition数据复制到多个server上,任何一个partition有一个leader和多个follower(可以没有);备份的个数可以通过broker配置文件来设定。leader处理所有的read-write请求,follower需要和leader保持同步.Follower就像一个"consumer",消费消息并保存在本地日志中;leader负责跟踪所有的follower状态,如果follower"落后"太多或者失效,leader将会把它从replicas同步列表中删除.当所有的follower都将一条消息保存成功,此消息才被认为是"committed",那么此时consumer才能消费它,这种同步策略,就要求follower和leader之间必须具有良好的网络环境.即使只有一个replicas实例存活,仍然可以保证消息的正常发送和接收,只要zookeeper集群存活即可.

选择follower时需要兼顾一个问题,就是新leader server上所已经承载的partition leader的个数,如果一个server上有过多的partition leader,意味着此server将承受着更多的IO压力.在选举新leader,需要考虑到"负载均衡",partition leader较少的broker将会更有可能成为新的leader.

[https_sookocheff.com_post_kafka_kafka-in-a-nutshell]: https://sookocheff.com/post/kafka/kafka-in-a-nutshell/
[7f50ac0f301e32cb70c0d75cbacfb1ae.png]: /images/20221005/521b1671988e403aaa14aae644943501.png
[3b49a49c726d53a1d6d90498f75ecd27.png]: /images/20221005/8f7771b3fa624651a20db870d26029cd.png
[f721d7bd05789f4aabebd2a5b69330e7.png]: /images/20221005/d89c2e6f350243b6810ebb2f051e879d.png
[log-anatomy.png]: /images/20221005/606e67b371c446b6a528b3e915b317d0.png
[da16f907e8ba5f8560bdee21b3873e51.png]: /images/20221005/f3a9ca06a27843a7a16176f1276694a6.png
[.png]: /images/20221005/9549b929615d49c7a30a5e7d9581af4f.png
[.png 1]: /images/20221005/08a3a263717c41aa96376994a6b6ead2.png
[.png 2]: /images/20221005/621b30e9a50e4db39588781dec8115bd.png
[Link 1]: http://kafka.apache.org/documentation.html
[.png 3]: /images/20221005/9e0448491b00447ca1253cfc3a43576f.png
[https_www.cnblogs.com_dreamroute_p_13092117.html]: https://www.cnblogs.com/dreamroute/p/13092117.html
[Link 2]: https://cloud.tencent.com/product/cfs?from=10680
[Hbase]: https://cloud.tencent.com/product/hbase?from=10680
[Link 3]: https://cloud.tencent.com/product/clb?from=10680