kafka学习

「爱情、让人受尽委屈。」 2023-10-06 18:44 57阅读 0赞

## 一、概述 ##

Kafka 的核心功能是什么？一言以蔽之，高性能的消息发送与高性能的消息消费。

Kafka 的设计初衷就是为了解决互联网公司超大数量级数据的实时传输。为了实现这个目标， Kafka 在设计之初就需要考虑以下4个方面的问题。

*  吞吐量／延时
 *  消息持久化
 *  负载均衡和故障转移
 *  伸缩性

#### 1.吞吐量/延迟 ####

kafka可以采取批量poll数据的方式提高吞吐量，或者合理的加分区。虽然kafka的数据最终是持久化到磁盘上的，但本质上每次写入操作都是写入操作系统的页缓存（page cache）中，由操作系统决定何时讲页缓存回写到磁盘。

同事kakfa写入操作采取的是追加（append）的方式，只能在日志文件末尾追加写入新消息，避免了磁盘的随机写操作。

kafka的读消息会先经过页缓存，缓存命中便把页缓存的内存直接发送到网络的socket上。利用Linux平台的sendfile系统调用做到的。零拷贝技术。

总结：

*  大量使用操作系统页缓存，内存操作速度快且命中率高。
 *  Kafka 不直接参与物理 I/0 操作，而是交由最擅长此事的操作系统来完成。
 *  采用追加写入方式，摒弃了缓慢的磁盘随机读／写操作。
 *  使用以 sendfile 为代表的零拷贝技术加强网络间的数据传输效率。

#### 2.持久化 ####

Kafka 实现持久化的设计也有新颖之处。普通的系统在实现持久化时可能会先尽量使用内存，当内存资源耗尽时，再一次性地把数据回写到磁盘；

而Kafka 则反其道而行之，所有数据都会立即被写入文件系统的持久化日志中，之后 Kafka 服务器才会返回结果给客户端通知它们消息已被成功写入。

这样做既实时保存了数据，又减少了 Kafka 程序对于内存的消耗， 从而将节省出的内存留给页缓存使用，更进一步地提升了整体性能。

#### 3.负载均衡和故障转移 ####

负载均衡由分区的leader实现。集群中的所有机器的分区都有机会分散leader实现负载均衡。

故障转移维持和zookeeper的会话，会话超时会选举新的机器提供服务

#### 4.伸缩性 ####

kafka服务器的状态统一由zookeeper管理扩展 Kafka 集群也只需要一步 ：启动新的 Kafka 服务器即可。当然这里需要言明的是，

在Kafka 服务器上并不是所有状态都不保存，它只保存了很轻量级的内部状态，因此在整个集群间维护状态 致性的代价是很低的。

## 二、kafka的术语 ##

#### 1.broker ####

一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。可通过配置broker.id指定。

#### 2.topic（主题）和partition（分区） ####

topic是一个逻辑概念，它代表一类消息。通常使用topic来区分实际业务，不同的业务对应不同的topic。而kafka的topic通常会被多个消费者订阅，

kafka并不是topic-message的两级结构，而是topic-partition-message的三级结构来分散负载。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA_size_16_color_FFFFFF_t_70][]

#### 3.offset（偏移量） ####

每个partition是不可修改的有序消息序列，offset表面当前的位置，消息在单分区是有序的，不保证全局有序。

通过<topic,partition,offset>的三元组可以在kafka集群中找到唯一对应的消息。

#### 4.replica（副本） ####

replica可以理解为kafka对消息的备份，目的就是防止数据丢失。副本分类两种：leader 和 follower，比较类似于MySQL的主备，但是有区别。

只有leader是可以对外提供服务的，follower只是被动的同步leader的数据。一旦leader节点故障会选举新的leader接替它的工作。

Kafka 保证同一个 partition 的多个 replica 一定不会分配在同一台 broker 。毕竟如果同一个broker 上有同一个 partition 的多个 replica ，那么将无法实现备份冗余的效果。

#### 5.ISR ####

Kafka partition 动态维护一个 replica 集合。该集合中的所有 replica 保存的消息日志都与 leader replica 保持同步状态。只有这个集合中的 replica 才能被选举为 leader ，

也只有该集合中所有 replica 都接收到了同一条消息， Kafka 才会将该消息置于“己提交”状态，即认为这条消息发送成功。

## 三、producer ##

#### 1.producer的工作流程 ####

producer 首先使用一个线程（用户主线程，也就是用户启动 producer 的线程）将待发送的消息封装进一个ProducerRecord 类实例，然后将其序列化之后发送给 partitioner（分区选择器）

再由后者确定了目标分区后一同发送到位于 producer 程序中的一块内存缓冲区中。producer的另一个工作线程。（I/O发送线程，也称 Sender 线程）则负责实时地从该缓冲区中提取出准备就绪的消息封装进一个批次（ batch ），统一发送给对应的 broker。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA_size_16_color_FFFFFF_t_70 1][]

#### 2.分区策略 ####

Kafka producer 发送过程中一个很重要的步骤就是要确定将消息发送到指定的 topic 的哪个分区中。 producer 提供了分区策略以及对应的分区器（partitioner）供用户使用。

Kafka默认的 partitioner 会尽力确保具有相同 key 的所有消息都会被发送到相同的分区上（采取hash取模的方式）；若没有为消息指定 key ，则该 partitioner 会选择轮询的方式来确保消息在 topic 的所有分区上均匀分配。

用户也可以自定义分区策略，根据需求指定不同类型的消息进入对应的分区，在配置props的时候指向自定义分区的实现即可。

## 四、consumer ##

#### 1.订阅topic ####

**订阅code** 展开源码

#### 2.offset管理 ####

kafka默认是自动提交位移的，自动间隔时间5s，可通过 a[uto.commit.interval.ms][] 参数控制提交间隔。

自动位移提交的优势是降低了用户的开发成本使得用户不必亲自处理位移提交：劣势是用户不能细粒度地处理位移的提交，特别是在有较强的精确一次处理语义时。在这种情况下，用户可以使用手动位移提交。  
所谓的手动位移提交就是用户自行确定消息何时被真正处理完并可以提交位移。在consumer 应用场景中，用户需要对 poll 方法返回的消息集合中的消息执行业务级的处理。 用户想要确保只有消息被真正处理完成后再提交位移。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA_size_16_color_FFFFFF_t_70 2][]

#### 3.rebalance ####

rebalance 本质上是一组协议，它规定了一个 consumer group 是如何达成一致来分配订阅 topic 的所有分区的。

rebalance 触发的条件

*  组成员发生变更，比如新 consumer 加入组，或己有 consumer 主动离开组，再或是己有consumer 崩溃时则触发 rebalance
 *  组订阅 topic 数发生变更，比如使用基于正则表达式的订阅，当匹配正则表达式的新 topic 被创建时则会触发 rebalance
 *  组订阅 topic 的分区数发生变更，比如使用命令行脚本增加了订阅 topic 的分区数

## 五、实现精确一次处理语义 ##

#### 1.消息的交付语义 ####

*  最多一次（at most once ）：消息可能丢失也可能被处理，但最多只会被处理一次。
 *  至少一次（at least once ）：消息不会丢失，但可能被处理多次
 *  精确一次（exactly once ）：消息被处理且只会被处理一次。

三种交付语音都有各自的应用场景，没有好坏之分。具体使用哪个要根据业务场景来确定。

在0.11.0.0版本之前，kafka producer默认提供at least once语义。设想当producer发送消息后，分区leader副本所在的broker成功将消息写入磁盘后给producer回执。若中间由于网络原因producer未收到回执

producer认为这条消息没有发送成功会进行重试，若重试后网络正常，那么一条消息会被写入磁盘两次。0.11.0.0版本引入了幂等性和事务的支持，解决了消息重复发送的问题。

#### 2.幂等性 ####

引入幂等性解决的就是瞬时发送错误导致的producer重试可引起的消息重复问题，开启幂等用户需要显式地设置 producer 端的参数 enable.idempotence 为 true。保证发送单分区的消息只会发送一次。

幂等的实现类似于TCP的工作方式，每次发送都会生成一个sequence number用于去重，kafka会把序列号保存到日志中，以保证即使有leader不可用，新的leader可以继续去重消息。

除了序列号， Kafka 还会为每个 producer 实例分配 producer id （称 PID ）。 producer 在初始化时必须分配一个 PID ，消息要被发送到的每个分区都有对应的序列号值，它们总是从0开始并且严格单调增加的。

对于 PID 、分区和序列号的关系，用户可以设想 Map, key 就是（PID 分区号）, value 就是序列号。即每对（PID ，分区号）都有对应的序列号值。若发送消息的序列号小于或等于broker 端保存的序列号，那么 broker 会拒绝这条消息的写入操作。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA_size_16_color_FFFFFF_t_70 3][]

#### 3.事务 ####

kafka的事务必须由用户显示的指定事务id

**kafka事务配置** 折叠源码

<table> 
 <tbody> 
 <tr> 
 <td> <code>@Bean</code> <code>public</code>&nbsp;<code>ProducerFactory&lt;String, String&gt; transactionalProducerFactory() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>DefaultKafkaProducerFactory factory =&nbsp;</code><code>new</code>&nbsp;<code>DefaultKafkaProducerFactory&lt;&gt;(producerConfigs());</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 开启事务</code> <code>&nbsp;</code><code>factory.transactionCapable();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 用来生成Transactional.id的前缀</code> <code>&nbsp;</code><code>factory.setTransactionIdPrefix(</code><code>"tran-"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>factory;</code> <code>}</code> <code>@Bean</code><code>(name =&nbsp;</code><code>"transactionalTemplate"</code><code>)</code> <code>public</code>&nbsp;<code>KafkaTemplate transactionalTemplate() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>new</code>&nbsp;<code>KafkaTemplate&lt;&gt;(transactionalProducerFactory());</code> <code>}</code> <code>使用 transactionalTemplate 发送消息时就可以使用</code><code>@Transactional</code><code>注解开启事务</code> </td> 
 </tr> 
 </tbody> 
</table>

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210728195733160.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210728195748602.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA==,size_16,color_FFFFFF,t_70
[uto.commit.interval.ms]: http://uto.commit.interval.ms/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/2021072819561879.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20210728195553450.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xpWmhlbjMxNA==,size_16,color_FFFFFF,t_70