Kafka原理

爱被打了一巴掌 2021-05-18 20:36 578阅读 0赞

kafka是LinkedIn公司开发的一种分布式的基于“发布/订阅”的消息中间件。

**1、主要特点：**

1、发布/订阅高吞吐量。其中它的消息持久化的访问时间复杂度为O(1)，即使对TB级别的数据也能维持常数时间的访问。

2、消息持久化。

3、分布式。易于向外拓展

4、消息消费采用拉模式（pull）。

5、支持Online和OffLine，同时支持离线数据处理和实时数据处理。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70][]

上图是一个典型的kafka集群。多个生产者将消息以“推模式”发布到Broker(kafka集群中的服务器)，多个消费者从kafka集群中以“拉模式“消费消息。同时通过Zookeeper对集群进行管理，选举leader，消费者改变时的再均衡（rebalance）。

**推模式(push)：在这里就是生产者主动将消息推送给kafka集群**

**拉模式(pull)：消费者主动从kafka队列中拉取消息。当消费者消费能力小于生产者生产能力时，可能产生消息堆积，延迟增大等现象。**

**Topic：**

Topic在逻辑上可以被认为是一个或一组queue，每条消费都必须指定它的Topic，可以简单理解为必须指明把这条消息放进哪个queue里。

为了使得Kafka的吞吐率可以线性提高，物理上把Topic分成一个或多个Partition（分区），每个Partition都是一个有序的队列，在物理上对应一个文件夹，该文件夹下存储这个Partition的所有消息和索引文件。

**Consumer Group：**

这是kafka用来实现一个Topic消息进行广播和单播的手段。它规定了一个topic消息可以有多个消费者组，但每个消费者组中只能有一个消费者进行消费。这样以后，如果要实现广播，只要一个消费者都对应一个独立的消费者组（1对1）；如果实现单播，只要所有消费者对应一个消费这组（所有 对1）。

**2、Topic && 分区**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70 1][]

一个Topic会有一个或多个分区，图中的partition就是分区，对应物理上是一个文件夹。一个topic的多个partition可能存在于不同的服务器上也可能存在于同一个服务器上的不同磁盘，这些分区在消息读写中可以并行处理。另外每个partition又会对应一个或多个segment，segment是物理存储上的最小单位，它由一个记录消息的数据文件和对应的索引文件构成。虽然会对应多个segment，但同一个partition的多个segment之间的读写不能并行处理，一个partition就像是一个很长的数组，是通过一个偏移量offset进行访问的。

**3、复制**

复制功能可以说是kafka的核心之一，它是一种高可用的解决方案。虽然有kafka集群功能，但当一台kafka服务器宕机的时候，它上面的分区就不可用了，这不是一种高可用的方案。所以通过zookeeper对kafka集群进行管理，选举leader和follower，客户端所有的请求都会先经过leader，follower的作用是从leader上复制消息的副本，当leader宕机的时候，zookeeper又会重新选举一个leader对外提供服务。

可以在server.properties中设置复制因子：

#设置复制因子,如果为3，表示按照一个leader，两个follower进行复制
    default.replication.factor=3

**4、消息发送**

**消息发送方式**

以java为例，虽然生产者都是通过Kafka-Client里面的Producer的send()方法进行发送消息，但也有三种情况：

1、立即发送：只需要将消息发给服务端，不管发送结果

2、同步发送：调用send方法发送消息后，会返回一个furture对象，根据furture对象判断消息是否发送成功。

3、异步发送：为send方法注册一个回调函数，这样当生产者消息发送成功接收到服务器响应后会触发回调函数。

**消息确认方式：**

上面说的消息发送方式是以生产者角度，消息确认是以服务器Broker角度。在“复制”一节中，我们说到了kafka集群采用leader、follower实现高可用方案，消息确认方式和这有关,主要也是三种：

1、不等broker响应，直接认为发送成功。这种方法效率高，但无法保证消息一定成功发送到服务器。

2、等leader响应，leader说消息发送成功了，就认为消息发送成功。但无法保证消息是否成功赋值到follower上。

3、等leader和所有的follower响应，大家都说成功了才认为消息发送成功，这是最稳的方式，但会耗费性能。

可见第二种是一种比较折中的方案。消息确认方式的初始化是在生产者初始化属性中的acks属性，属性值为0，对应第一种方法，1对应第二种，all对应第三种。

**消息重发：**

如果消息发送失败，生产者会选择重发消息。（猜想：在选择第一种消息确认方式后，消息发送总是成功，也就不存在消息重发了）。

主要有两个属性：重发次数和每次重发的时间间隔。

#重试次数
    retries=10
    #重试间隔,单位ms
    retry.backoff.ms=1000

**批次发送**

当有多条消息要发送到同一个分区，生产者会选择批次发送。就像公交车发车一样，不可能来一个人就开一趟（对应消息提交），会等待一会，等多几个人再发一次车，但也不会无休止等待。这样可以减少提交次数，减少服务器压力，提高吞吐量。

在上面公交车的例子中，有两种情况：一种是车上人坐满了，那么等待也没有意义了，直接发车。第二种是时间到了，但车上还没坐满，那也只能发车了。同理，这里有两个参数：

#批次大小，对应公交车座位数
    batch.size=16384
    #等待时间，对应公交车等待时间
    linger.ms=10

**消费者组：**

关于消费者组实现广播与单播的问题前面已经说过了。下面说的是当消费者组里面的消费者变更的问题。

我们知道，消费者组里面的消费者有可能增加，也有可能离开，在这种情况下，原来的消费者拥有的分区将会被让出来分配给其他消费者，这种分区所有权转变的过程叫做“再均衡”（Rebalance）。当消费者变更、订阅的主题数量变更、订阅主题的分区数变更 均会触发再均衡，再均衡是一个很重要的设计，但我们又不希望发生。因为再均衡的过程中，整个消费组的会暂停消费，直至再均衡结束。就像JVM中full gc的stop the world！

**消息偏移量Offset：**

前面说到，分区就像是一个大数组，消息的消费是通过offset偏移量（就像数组的索引）进行定位的。那么kafka是怎么记录每个分区当前消费到哪个位置了呢？

其实kafka是通过建立一个“\_consumer\_offset”主题来记录当前消费的偏移量的。消费者在消费消息后，都会往“\_consumer\_offset”主题中发送一条消息，消息为分区的偏移量，就像一个日志信息。当消费者正常运行，这个日志作用不大，因为它知道当前的offset，但如果该消费die了，被另外的消费者A取代了，那消费者A如何知道上一个消费者消费到分区的哪个位置了？此时有了“\_consumer\_offset”主题，消费者A就会去该主题下查找当前分区的偏移量（就像查日志一样），然后继续消费。

说到这里，似乎继续消费的问题就解决了。实际上没有！你怎么知道前一个消费者提交的偏移量就是对的呢？即使前一个消费者是消费一条提交一次偏移量，但如果当他消费完一条消息，还没来得及提交偏移量就挂了呢？取代的消费者A就会取到上一个偏移量，该条消息就会被重复消费，会产生问题。

目前kafka有这样几种方案：

1、自动提交，默认每隔5s提交一次。这种默认方案显然这样是有问题的。

2、手动提交。关闭自动提交，在一轮消息或每条消息消费完后，手动调用commitSync方法，提交偏移量。这种方法稳了很多，但调用commitSync方法时同步的，会降低消费者的吞吐量，降低性能。

3、看看异步提交。commitAsync方法比起同步提交，不需要消费者同步等待，同时在发生错误时会一直重试。这挺好。但如果有两次提交一直在重试，一次提交的偏移量为500，一次为300,当服务器恢复的时候，500先提交成功了，300的把500的覆盖了，这样就有问题了。（我想这里是不是还可以加一个逻辑，因为偏移量肯定是递增的，重试中对重试的任务维护一个重试队列，按照分区、偏移量进行分类排序，这样就不会有这样的问题了，你们觉得呢？欢迎留言讨论）

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70]: /images/20210518/c84aa38fbd534ecbac4a64a2845a6c3d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70 1]: /images/20210518/8db62c27e38449f592e601ab2b8db4b2.png