MQ系列笔记—Kafka相关

向右看齐 2022-08-31 09:41 194阅读 0赞

### 这里写目录标题 ###

*   *  一、简介
     *  二、基本概念
     *   *  消息
         *  主题与分区
         *  生产者与消费者
         *   *  生产者
             *  消费者
         *  Brokers和集群
         *  偏移量
         *   *  自动提交偏移量
             *  如何防止重复消费信息？
             *  如何保证消息不丢失?
         *  Kafka rebalance机制
         *  Springboot 集成 Kafka
         *   *  发送消息
             *  消费消息
         *  Kafka阻塞事故
         *  拾遗
     *  题外话
     *   *   *  参考

## 一、简介 ##

ApacheKafka 是一个分布式的流处理平台。它具有以下特点：

*  支持消息的发布和订阅，类似于 RabbtMQ、ActiveMQ 等消息队列；
 *  支持数据实时处理；
 *  能保证消息的可靠性投递；
 *  支持消息的持久化存储，并通过多副本分布式的存储方案来保证消息的容错；
 *  高吞吐率，单 Broker 可以轻松处理数千个分区以及每秒百万级的消息量。

## 二、基本概念 ##

### 消息 ###

Kafka 的基本数据单元被称为 `message(消息)`，为减少网络开销，提高效率，多个消息会被放入同一批次 (Batch) 中后再写入。

### 主题与分区 ###

Kafka 的消息通过 Topics(主题) 进行分类，一个主题可以被分为若干个 Partitions(分区)，一个分区就是一个提交日志 (commit log)。消息以追加的方式写入分区，然后以先入先出的顺序读取。Kafka 通过分区来实现数据的冗余和伸缩性，分区可以分布在不同的服务器上，这意味着一个 Topic 可以横跨多个服务器，以提供比单个服务器更强大的性能。

由于一个 Topic 包含多个分区，因此无法在整个 Topic 范围内保证消息的顺序性，但可以保证消息在单个分区内的顺序性。

![kafka-topic.png][]

### 生产者与消费者 ###

#### 生产者 ####

生产者负责创建消息。

一般情况下，生产者在把消息`均衡`地分布到在主题的所有分区上，而并不关心消息会被写到哪个分区。如果我们想要把消息写到指定的分区，可以通过自定义分区器来实现。

#### 消费者 ####

消费者是消费者群组的一部分，消费者负责消费消息，之所以有消费者组这一个概念，简单的讲，是因为单一消费者可能无法跟上数据生成的速度，此时可以增加更多的消费者，让他们分担，分别处理部分分区的消息，这也是Kafka`横向伸缩`的主要手段。

消费者可以订阅一个或者多个主题，并按照消息生成的顺序来读取它们。

消费者通过检查消息的`偏移量 (offset)` 来区分读取过的消息。

偏移量是一个不断递增的数值，在创建消息时，Kafka 会把它添加到其中，在给定的分区里，每个消息的偏移量都是唯一的。消费者把每个分区最后读取的偏移量保存在 Zookeeper 或 Kafka 上，如果消费者关闭或者重启，它还可以重新获取该偏移量，以保证读取状态不会丢失。

![kafka-producer-consumer.png][]

一个分区只能被同一个消费者群组里面的一个消费者读取，但可以被不同消费者群组中所组成的多个消费者共同读取。多个消费者群组中消费者共同读取同一个主题时，彼此之间互不影响。

![kafka_.png][]

### Brokers和集群 ###

一个独立的 Kafka 服务器被称为 Broker。

Broker 接收来自生产者的消息，为消息设置偏移量，并提交消息到磁盘保存。Broker 为消费者提供服务，对读取分区的请求做出响应，返回已经提交到磁盘的消息。

Broker 是集群 (Cluster) 的组成部分。每一个集群都会选举出一个 Broker 作为集群控制器 (Controller)，集群控制器负责管理工作，包括将分区分配给 Broker 和监控 Broker。

在集群中，一个分区 (Partition) 从属一个 Broker，该 Broker 被称为分区的首领 (Leader)。一个分区可以分配给多个 Brokers，这个时候会发生分区复制。这种复制机制为分区提供了消息冗余，如果有一个 Broker 失效，其他 Broker 可以接管领导权。

![kafka-cluster.png][]

整体来看也是这样的层级,Topic可对应多个分区(Partitions)  
![在这里插入图片描述][6aa72430993c4ac881b8a92d40df51c3.png]

### 偏移量 ###

Kafka 的每一条消息都有一个偏移量属性，记录了其在分区中的位置，偏移量是一个单调递增的整数。消费者通过往一个叫作 `＿consumer_offset` 的特殊主题发送消息，消息里包含每个分区的偏移量。 如果消费者一直处于运行状态，那么偏移量就没有  
什么用处。不过，如果有消费者退出或者新分区加入，此时就会触发再均衡。完成再均衡之后，每个消费者可能分配到新的分区，而不是之前处理的那个。为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的地方继续处理。 因为这个原因，所以如果不能正确提交偏移量，就可能会导致数据丢失或者重复出现消费，比如下面情况：

*  如果提交的偏移量小于客户端处理的最后一个消息的偏移量 ，那么处于两个偏移量之间的消息就会被重复消费；
 *  如果提交的偏移量大于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失。

Kafka 支持自动提交和手动提交偏移量两种方式。

#### 自动提交偏移量 ####

**enable.auto.commit**

是否自动提交偏移量，默认值是 true。为了避免出现重复消费和数据丢失，可以把它设置为 false。

使用自动提交是存在隐患的，假设我们使用默认的 5s 提交时间间隔，在最近一次提交之后的 3s 发生了再均衡，再均衡之后，消费者从最后一次提交的偏移量位置开始读取消息。这个时候偏移量已经落后了 3s ，所以在这 3s 内到达的消息会被`重复处理`。可以通过修改提交时间间隔来更频繁地提交偏移量，减小可能出现重复消息的时间窗，不过这种情况是无法完全避免的。基于这个原因，Kafka 也提供了手动提交偏移量的 API，使得用户可以更为灵活的提交偏移量。

#### 如何防止重复消费信息？ ####

有了上面偏移量的概念，我们就可以考虑重复消费的场景了，因为offset是在一定时间间隔后提交的，所以当系统重启时，可能还没有提交offset，即已经消费过了，就会出现重复消费消息的问题。

*  根据具体业务，保证幂等性，如果对`sql`数据库操作，比如插入操作，那么就判断是否已经存在，若已经存在，则不插入，或者更新都可以，而如果操作的是像`redis`这样的数据库，是天然保证幂等性的。
 *  生产方可以在发送消息的时候带上一个全局唯一ID，当我们消费后，把ID存到redis，后续再去拿redis，判断是否消费过

#### 如何保证消息不丢失? ####

*  **保证生产者数据不丢失**:`ACK`机制,在kafka发送数据的时候，每次发送消息都会有一个确认反馈机制，确保消息正常的能够被收到.可以设置ack,当`ack=0`时,producer`不等待broker`同步完成的确认，继续发送下一条(批)信息,当`ack=1`时（默认）,producer要等待`leader`成功收到数据并得到确认，才发送下一条message。当`ack=-1`时,producer得到`follwer`确认，才发送下一条数据。
 *  **保证消费者数据不丢失**:与防止重复消费数据的情景是类似的,使用offset机制,每过一会自动提交`commit`,虽然可能会出现重复消费,但不会出现丢失消息.
 *  **保证broker数据不丢失**:每个broker中的partition我们一般都会设置有replication（副本）的个数

### Kafka rebalance机制 ###

*  有新的消费者加入
 *  有消费者宕机或者下线
 *  消费者主动退出消费者组
 *  消费者组订阅的topic出现分区数量变化
 *  消费者调用`unSubscrible`取消对某topic的订阅

### Springboot 集成 Kafka ###

<dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
            </dependency>

#### 发送消息 ####

可以直接使用`KafkaTemplate`发送消息

@Autowired
    KafkaTemplate kafkaTemplate;

#### 消费消息 ####

使用监听器监听即可

@KafkaListener(groupId = "task-hr", topics = {
        HR_PERSON_TOPIC})

### Kafka阻塞事故 ###

Kafka在公司项目中用于传递消息，但有一天他阻塞了。。。

这事不怪Kafka，是人为原因

为了保证事务，我们可以使用`@Transactional`注解来实现事务管理，有人在Service上加了`@Transactional`，当执行成功，才算从Kafka中消费了该条消息，但这就导致作用域太大，如果Service中有十条是update，有一条因为某种原因失效了，那么就会回滚，从而一直堵塞在这里。。。

### 拾遗 ###

通过EFK平台查日志，发现向ActiveMQ发送消息花了140s，woc，考虑是同步发消息，消息多了就慢，但是查官网发现ActiveMQ是默认异步，随即开始找别的原因。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzg4OTg0MQ_size_16_color_FFFFFF_t_70]  
后来debug发现，项目ActiveMQ版本较为老，并非默认异步，设置异步后，速度提升到1ms,好在消息可以容忍丢失，异步体验超好

## 题外话 ##

2021.9月，Apache Kafka 3.0 发布！

主要的更新包含以下几个点：

*  Apache Kafka 的内置共识机制将取代 Apache ZooKeeper
 *  从 Apache Kafka 3.0 开始，生产者默认启用最强的交付保证 ( acks=all, enable.idempotence=true)。这意味着用户现在默认获得排序和持久性。
 *  弃用 Kafka 中对 Java 8 的支持

#### 参考 ####

http://activemq.apache.org/async-sends.html

[kafka-topic.png]: /images/20220829/d3f9c58f874d413886175de55751c898.png
[kafka-producer-consumer.png]: /images/20220829/6cd71f125f674ce8b177d76378b7e984.png
[kafka_.png]: /images/20220829/3b73dceaf7cc4a9b9308d1f2e61bb832.png
[kafka-cluster.png]: /images/20220829/33acf84505384a499652fb37fbd5a879.png
[6aa72430993c4ac881b8a92d40df51c3.png]: /images/20220829/0515a26a6e3f4df79fd18e938e267715.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzg4OTg0MQ_size_16_color_FFFFFF_t_70]: /images/20220829/f4a4d73f47d8449fa9a4c5d340ad522e.png