一、概述

（一）、kafka的定义

1、定义

1）kafka传统的定义：kafka是一个分布式的基于发布/订阅模式的消息队列，主要用于大数据实时处理领域

2）kafka最新的定义：kafka是一个开源的分布式事件流平台（event stream platform），主要用高性能数据管道，流分析，数据集成和关键任务等领域

2、消息队列

目前市面上大部分公司采用的消息队列主要有kafka，activeMQ，rabbitMQ，rocketMQ等。kafka作为消息队列，主要应用于大数据场景下，而在Javaee开发中更多采用的是activeMQ，rabbitMQ,rockectMQ等。

3、消息队列的应用场景

传统的消息队列的主要应用场景包括：缓冲/削峰，解耦和异步通信

缓冲/削峰：在实际的应用系统中，如果数据生产端（比如其前端）的数据产生的速率与数据处理端（服务端）的数据处理速率相当或小于时，整合系统运行就不会有很大的压力。但是当系统上了个秒杀活动或者双11活动到来，前端用户猛增，数据率随之也会增加数倍，甚至是数十倍。但是服务端需要对数据进行处理，持久化等操作，处理速率必然跟不上数据产生的速度，久而久之系统就会产生数据积压，最终就有可能导致系统的崩溃。在数据生产端和处理端之间使用消息队列就可以解决这种问题。此时，消息队列就发挥了不同系统之间数据的缓冲和削峰的作用。数据生产端将数据发送到消息队列，然后随即返回响应，这个过程相对来说是非常快的。数据处理端则根据自己的处理速度从消息队列中拉取数据。示意图如下：

image

不使用消息队列的情况

image

使用消息队列的情况

解耦：允许独立的拓展和修改两边的处理过程，但两边需要确保使用相同的接口约束。

image

异步通信：将处理的用户数据写入到消息队列中，并立即返回处理结果，队列中数据由另一个线程拉取出来做响应的处理。下面是用户注册，并把注册成功的消息发送到用户手机上的同步处理和异步处理的流程。

image

（二）、kafka基础架构

1、消息队列的两种模式

　1）点对点模式

                           消费者主动拉取数据，数据消费完后就会在队列中删除

2）发布/订阅模式

             可以有有多个主题（topic）
              消费者拉取数据消费完后，不删除数据
             每个消费者相互独立，都可以消费到数据　　

2、基础架构

image

1）producer：消息生产者，就是向broker发送消息的客户端

2）consumer：消息消费者，就是从broker拉取数据的客户端

3）consumer group：消费者组，由多个消费者consumer组成。消费者组内每个消费者负责消费不同的分区，一个分区只能由同一个消费者组内的一个消费者消费；消费者组之间相互独立，互不影响。所有的消费者都属于某个消费者组，即消费者组是一个逻辑上的订阅者。

4）broker：一台服务器就是一个broker，一个集群由多个broker组成，一个broker可以有多个topic。

5）topic：可以理解为一个队列，所有的生产者和消费者都是面向topic的。

6）partition：分区，kafka中的topic为了提高拓展性和实现高可用而将它分布到不同的broker中，一个topic可以分为多个partition，每个partition都是有序的，即消息发送到队列的顺序跟消费时拉取到的顺序是一致的。

7）replication：副本。一个topic对应的分区partition可以有多个副本，多个副本中只有一个为leader，其余的为follower。为了保证数据的高可用性，leader和follower会尽量均匀的分布在各个broker中，避免了leader所在的服务器宕机而导致topic不可用的问题。

8）leader：多个副本的主副本，生产者发送的数据和消费者消费的数据都是通过leader进行处理的。

9）follower：多个副本中除了leader副本，其余的均为follower副本，也即从副本。从副本不会和生产者和消费者提供服务，而是实时同步主副本的数据。当主副本宕机后，通过一定算法选举出新的从副本成为主副本，继续为生产者和消费者提供服务。

（三）、kafka常用命令行操作

1、主题相关

| 参数 | 描述 |
| --bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号。 |
| --topic <String: topic> | 操作的 topic 名称。 |
| --create | 创建主题。 |
| --delete | 删除主题。 |
| --alter | 修改主题。 |
| --list | 查看所有主题。 |
| --describe | 查看主题详细描述。 |
| --partitions <Integer: # of partitions> | 设置分区数。 |
| --replication-factor<Integer: replication factor> | 设置分区副本。 |
| --config <String: name=value> | 更新系统默认的配置。 |

2、生产者相关

| **参数 ** | 描述 |
| --bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号。 |
| --topic <String: topic> | 操作的 topic 名称。 |

3、消费者相关

| **参数 ** | 描述 |
| --bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号。 |
| --topic <String: topic> | 操作的 topic 名称。 |
| --from-beginning | 从头开始消费。 |
| --group <String: consumer group id> | 指定消费者组名称。 |

二、生产者

（一）、重要参数列表

|

参数名称

| 描述 |
|

bootstrap.servers

| 生产者连接集群所需的 broker 地 址 清 单 。可以设置 1 个或者多个，中间用逗号隔开。注意这里并非需要所有的 broker 地址，因为生产者从给定的 broker里查找到其他 broker 信息。 |
|

key.serializer 和 value.serializer

| 指定发送消息的 key 和 value 的序列化类型。一定要写全类名。 |
|

buffer.memory

| RecordAccumulator 缓冲区总大小，默认 32m。 |
|

batch.size

| 缓冲区一批数据最大值，默认 16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。 |
|

linger.ms

| 如果数据迟迟未达到 batch.size，sender 等待 linger.time之后就会发送数据。单位 ms，默认值是 0ms，表示没有延迟。生产环境建议该值大小为 5-100ms 之间。 |
|

acks

|

• 0：生产者发送过来的数据，不需要等数据落盘应答。
• 1：生产者发送过来的数据，Leader 收到数据后应答。
• -1（all）：生产者发送过来的数据，Leader+和 isr 队列里面的所有节点收齐数据后应答。默认值是-1，-1 和all 是等价的。

|
| max.in.flight.requests.per.connection |

允许最多没有返回 ack 的次数，默认为 5，开启幂等性要保证该值是 1-5 的数字。

|
|

retries

|

当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是 int 最大值，2147483647。如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功了。

|
|

retry.backoff.ms

|

两次重试之间的时间间隔，默认是 100ms。

|
|

enable.idempotence

|

• 是否开启幂等性，默认 true，开启幂等性。*

|
|

compression.type

|

• 生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩。支持压缩类型：none、gzip、snappy、lz4 和 zstd。*

|

（二）、发送流程以及发送API

1、发送流程

image

（三）、分区

1、分区的好处

• • • 便于合理使用存储资源。每一个partition存储在不同的broker上面，可以把海量的数据切割成更小数据存储在不同的broker上。合理是分配分区的任务，可以实现负载均衡效果，避免单机数据量太大而导致的压力。
    • 提高并行度。生产者可以以分区为单位发送数据；消费者可以以分区为单位消费数据。
      image

2、分区策略

1）、默认分区策略：DefaultPartitioner

• • • 若生产者发送数据时指明partition，那么就会将该数据发送指定的分区，比如partition=0，则数据发送到第零个分区上
    • 若没有指明partition，但是有key的情况下，用key的hashcode值与该topic对应的partition的数量进行取余得到partition，然后将数据发送到该分区上。比如：key的hashcode值为5，topic对应的分区数为3，那么 5 % 3 = 2，所以该数据应该发送到第二个分区。
    • 既没有指定partition和key，kafka采用粘性分区器，会随机选择一个分区，然后尽可能使用该分区，指导该分区batch-size满了或已完成，然后再随机寻找另一个分区（与当前的分区不一样）。比如：这次随机到分区0，等当前批次满了或者linger.ms时间已到，kafka再随机一个分区，如果仍然随机到分区0则继续随机。

　　　　　　2）自定义分区器

　　　　　　　　在实际的企业应用中，可能会有不用的场景，默认的分区器无法满足需求，那么就需要自定也分区器来满足需求，比如某个分区的服务器性能比较好，另一个是比较久的服务器，性能相对差一点，那么就需要通过自定义分区器让更多的数据向性能更好的分区倾斜。

　　　　　　　　自定义分区器的实现：第一，定义类实现 Partitioner 接口。第二，重写partition()方法。第三，在生产者配置中指定自定义的分区器。

（四）、提高生产者的吞吐量

要想更好的提高生产者的吞吐量，则必须先了解生产者发送数据的流程，具体流程可以看靠（一）的介绍。以下是提高生产者的吞吐量的建议。

1. 1. 修改batch.size的大小。batch.size默认大小为16k，提高批次大小可以一定程度提高吞吐量。这就好比用小卡车拉货物和用大卡车拉货物，小卡车单位时间内不能一次拉完货物，那么来回就需要消耗额外时间，而大卡车一次性把货物拉走，那么就节省了来回的时间。

   2. 修改linger.ms的时间，默认为0，即直接发送数据，一般设置为5-100ms，通过设置延迟时间一次性可以发送更多的数据。

   3. 修改compression.type的数据压缩类型，默认snappy。根据不同的业务场景选择不同的数据压缩方法，提高数据压缩率。kafka提供的压缩类型有：gzip,snappy,lz4,zstd。

   4. 修改RecordAccumulator缓冲区的大小。默认32m，增加缓冲区大小可以那么每个batch.size的值就更大，每次发送的数据更多。

（五）、数据可靠性

数据的可靠性是指producer发送数据到kafka收到应答后，该数据都能成功落盘，那么这次发送是可靠的。但是，为了适应不用的应用场景以及实现高可用，kafka会将数据备份到不同的副本当中，在数据同步的过程中如果出现的故障，那么就有可能出现数据丢失，重复情况。想要弄清楚kafka数据可靠性，就必须先要了解kafka中ACK的应答原理。

1、ACK应答原理

ACK应答是指在leader分区接收到生产者的数据后，何时对生产者做出应答的策略。ACK可选的值有0，1，-1三个，可以在生产者的配置项 acks 中设置，ACK设置不同，对生产者做出应答的时机也不同。

ACK=0：可靠性级别最低。leader收到生产者数据后不需要等数据落盘，立即对生产者做出应答。生产者收到应答后认为leader已成功接收数据，因此不需要再发当前数据了。但是，如果leader在将内存中的数据落盘时突然出现故障，那么这条数据因为没有保存到磁盘中而导致数据的丢失。

ACK=1：可靠性级别较高。leader收到生产者的数据并将数据落盘后，对生产者做出应答。生产者收到应答后继续发送其他数据。如果leader做出应答并且follower未同步到该数据时，leader出现故障。kafka会重新在follower中选出新的leader，而新的leader心有同步到数据，生产者也不会再发该数据，因此导致该数据的丢失。

ACK=-1（all）：可靠性级别最高，kafka的acks默认值。leader收到数据并落盘，并且确认所有follower收到数据后再给生产者应答。此时，所有分区副本都有该数据了，即使任意分区出现故障数据仍然是完整的。

（六）、数据去重

1、数据重复原因

在使用kafka时为了保证数据的高可靠性，我们一般都会将应答级别设置为-1（all），即leader的ISR列表的follower均收到数据后再应答。非常不幸的是，leader在收到所有的follower的确认后发生故障，所有的分区均已保存到磁盘中，但是生产者没有收到应答，认为leader没有收到生产者发送请求，于是尝试重新发送请求。由于leader发生故障，kafka重新选举leader，生产者将数据再一次发送到新的leader上，所以造成的数据重复。

image

2、数据去重

kafka 0.11之后引入了幂等性和事务两大特性。利用这两个特性可解决数据重复的问题。

1）数据传递语义

• • • • • 至少一次（At Least Once）：ACK级别为-1 + 分区副本大于等于2 + ISR中应答的最小副本数量大于等于2；
        • 至多一次（At Most Once）：ACK应答级别为 0；

At Least Once可以保证数据不丢失，但不能保证数据重复。

At Most Once 可以保证数据不重复，但不能保证数据不丢失。

2）幂等性的原理

幂等性：是指无论producer发送多少条重复的数据，broker端都只会持久化一条数据，保证了数据不重复。

数据重复的判断依据：具有<PID, Partition, SeqNumber>相同主键的消息提交时，Broker只会持久化一条。其中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的。

　　　　　　　　　　　　　　从数据重复判断依据来看，幂等性只能保证单分区会话内数据不重复。

开启幂等性：enable.idempotence = true,默认为true。

image

3）事务

kafka的事务需要跟幂等性配合起来使用。开启事务就必须开始幂等性。

image

（七）、数据乱序和有序

1、数据乱序

kafka的producer客户端在向broker发送数据时并不是直接将数据发送出去，而是将数据先缓存到本地的双端缓存队列中，sender线程会不断地检测缓存队列中地数据，若队列中地数据达到了设置的（batch.size）值地容量或者达到一定的时间（linger.ms），sender就会创建一个InFlightRequests线程，该线程负责将数据发送到broker上。kafka 默认的InFlightRequests是5个，InFlightRequests线程发送数据后不需要broker的应答就可以发下一个数据，因此最多可以一起发5个请求。比如，需要发送0，1，2，3，4这五个数据依次被发送出去，但是2这个数据没有接收成功，客户端则重新2的数据，此时broker接收到的数据的顺序就变为了01342，而非01234。

　　　　　　2、数据有序

　　　　　　　　1）kafka在1.x之前保证数据单分区有序，需要将InFlightRequests线程数设置为1个（****max.in.flight.requests.per.connection=1）。当线程数为1时就能保证broker收到数据确认后再发下一条数据。**

*****　　　　　　　　2）kafka在1.x以后在未开启幂等性的情况下，处理流程跟1.x以前的版本一样。*****

*****　　　　　　　　3）1.x以后在开启幂等性的情况下，可以将max.in.flight.requests.per.connection设置为小于等于5。原因是，Kafka服务端会缓存最近发过来的元数据，等缓存满了5个后就会对这些元数据进行排序，这样就可以保证数据有序了。*****

三、broker

（一）、工作流程

1、重要参数

| **参数名称 ** | 描述 |
| replica.lag.time.max.ms | ISR 中，如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值，默认 30s。 |
| auto.leader.rebalance.enable | 默认是 true。 自动 Leader Partition 平衡。 |
| leader.imbalance.per.broker.percentage | 默认是 10%。每个 broker 允许的不平衡的 leader的比率。如果每个 broker 超过了这个值，控制器会触发 leader 的平衡。 |
| leader.imbalance.check.interval.seconds | 默认值 300 秒。检查 leader 负载是否平衡的间隔时间。 |
| log.segment.bytes | Kafka 中 log 日志是分成一块块存储的，此配置是指 log 日志划分 成块的大小，默认值 1G。 |
| log.index.interval.bytes | 默认 4kb，kafka 里面每当写入了 4kb 大小的日志（.log），然后就往 index 文件里面记录一个索引。 |

2、zookeeper存储的kafka信息

image

3、工作的总流程

① broker启动，向zookeeper注册broker节点信息

② 向zookeeper注册controller，若注册成功则该controller负责选举leader，监听broker节点等工作

③ 注册成功的controller开始监听brokers节点变化

④ 注册成功的controller举行leader选举。按照AR列表优先选取排在前面的节点，若该节点在ISR中存活则成为leader，否则继续轮询。例如：AR[0,1,2],ISR[1,2]，controller会从broker 0 开始轮询，发现ISR中并没有0，那么继续轮询到1，发现ISR有1则1成功leader。

⑤ 注册成功的controller将选举结果写入到zookeeper中，同时其他controller也监听该信息，等leader故障后方便立即上位成为新的leader。

⑥ 加入leader挂掉了，controller监听到节点发送变化，开始拉取ISR信息，然后开始重新选择leader。

image

（二）、副本

1、基本信息

• • • 作用：提高数据可用性
    • Kafka默认的副本数为1，生产环境可设置为2，保证数据可靠性。不建议设置太多的副本，一者增加磁盘存储空间，二者增加副本同步时的网络开销，影响效率。
    • 副本分为 leader 和follower。leader只有一个，其余的为follower。生产者的数据发给的是leader，消费者消费的消息也是来自leader的。follower同步leader的消息。
    • AR：分区中所有副本的总和。
    • ISR：分区中能够和leader保持同步的副本，包括leader本身。如果follower长时间未向leader发送通信信息或同步请求，那么该follower就会从ISR中移除。leader挂了后ISR中的follower重新选举leader。
    • OSR：分区中跟leader同步信息差距较大的副本或者是挂了的副本。
    • AR = ISR + OSR

2、leader选举流程

Kafka集群中有一个会被选举为controller leader，选举方式抢占式，谁先抢占到zookeeper的节点，谁就能成为leader。该controller负责Kafka集群broker的上下线，topic分区副本的分配和leader选举等工作。

3、leader和follower故障处理

LEO（log end offset）：每个副本的最后一个offset，即最新offset + 1。

HW（high watermark）：所有副本中最小的LEO。

image

1）leader发生故障

leader发生故障后，kafka会从ISR中重新选举出新的leader

为保证多副本之前的数据一致性，其余的follower会先将自己高于HW的数据截掉，然后同步新leader的数据。

image

2) follower发生故障

follower发生故障后会被临时踢出ISR。

此时leader和follower继续接收数据。

等follower恢复后读取保存在磁盘的HW，并将log文件中高于HW部分的数据截取丢弃，从HW位置开始同步leader的数据。

follower同步到LEO大于等于分区的HW时，kafka就将该副本重新加入到ISR中。

4、副本分配

1）尽量将所有副本平均地分配到所有地broker上。

2）每个broker分配到地leader尽可能一样多。

3）leader和follower尽可能地分配到不同地broker上。

5、leader分区自平衡

正常情况下，Kafka本身会尽可能地将leader分区均匀地分布到各个机器上，这样使得每个机器的读写吞吐量比较均匀。但是，如果当leader分区故障后，另一个follower就会迅速上位并且承担之前的leader的工作，读写请求的压力就会上升，造成集群负载不平衡。kafka允许的每个broker不平衡的比率默认为10%，如果超过了这个值就会触发leader分区平衡。

相关参数：

| 参数名称 | 描述 |
| auto.leader.rebalance.enable | 默认是 true。 自动 Leader Partition 平衡。生产环境中，leader 重选举的代价比较大，可能会带来性能影响，建议设置为 false 关闭。 |
| leader.imbalance.per.broker.percentage | 默认是 10%。每个 broker 允许的不平衡的 leader的比率。如果每个 broker 超过了这个值，控制器会触发 leader 的平衡。 |
| leader.imbalance.check.interval.seconds | 默认值 300 秒。检查 leader 负载是否平衡的间隔时间。 |

（三）、数据存储

1、存储机制

Kafka中topic是逻辑概念，但是分区partition是物理概念，生产者发送的数据都是存储在partition中的，每个partition中都有一个log文件，数据实际都是存储在log文件中，生产者生产的数据都会最佳到该log文件的末端。为防止log文件过大导致数据检索过慢，kafka将log文件进行切片，每个片成为segment，每个segment中包括“.index”文件，“.log”文件和“.timestamp”等文件。

log文件：保存实际数据的文件。

index文件：偏移量索引文件，文件为segment第一个数据offset值。index文件是一个稀疏索引，每往log文件中写入4kb的数据才向index文件中写入一条索引。

timestamp文件：时间戳索引文件，文件名为segment第一条数据的offset值。

image

2、数据文件清楚策略

Kafka提供的文件删除策略有delete和compact两种。

delete策略：将过期数据删除

log.cleanup.plocy = delete 所有数据启用删除策略

• • • • 基于时间删除，默认打开。以segment中所有记录中最后记录的时间戳作为该文件的时间戳。
      • 基于大小删除，默认关闭。超过设置的所有日志总大小，删除最早的segment。

compact日志压缩：对于相同key的不同value值，只保留最后一个value值。压缩后的offset可能是不连续的。这种策略只适合某些特殊的场景，比如key保存的是用户id，value保存的用户信息，通过压缩策略就能将旧的用户数据删除，只保留新的用户信息。

（四）、高效读写数据

1、Kafka是基于分布式集群，采用分区技术，数据读写并行度高。

2、读数据采用稀疏索引，消费者能够快速定位消费的数据。

3、顺序写磁盘，producer生产数据只是在log文件的末端追加数据，因此写磁盘的速度很快。

4、Kafka以来系统底层的pagecache技术和零拷贝技术，能够提高磁盘数据读写速度。

四、消费者

（一）、重要参数

| 参数名称 | 描述 |
| bootstrap.servers | 向 Kafka 集群建立初始连接用到的 host/port 列表。 |
| key.deserializer 和 value.deserializer | 指定接收消息的 key 和 value 的反序列化类型。一定要写全
类名。 |
| group.id | 标记消费者所属的消费者组。 |
| enable.auto.commit | 默认值为 true，消费者会自动周期性地向服务器提交偏移量。 |
| auto.commit.interval.ms | 如果设置了 enable.auto.commit 的值为 true， 则该值定义了消费者偏移量向 Kafka 提交的频率，默认 5s。 |
| auto.offset.reset |

当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在如，数据被删除了），该如何处理？

earliest：自动重置偏移量到最早的偏移量。

latest：默认，自动重置偏移量为最新的偏移量。

none：如果消费组原来的（previous）偏移量存在，则向消费者抛异常。

anything：向消费者抛异常。

|
| offsets.topic.num.partitions | __consumer_offsets 的分区数，默认是 50 个分区。 |
| heartbeat.interval.ms | Kafka 消费者和 coordinator 之间的心跳时间，默认 3s。该条目的值必须小于 session.timeout.ms ，也不应该高于session.timeout.ms 的 1/3。 |
| session.timeout.ms | Kafka 消费者和 coordinator 之间连接超时时间，默认 45s。超过该值，该消费者被移除，消费者组执行再平衡。 |

（二）、消费方式

1、pull（拉）模式

consumer主动拉取服务端的数据。Kafka采用这种方式。

优点：consumer根据自己的处理能力拉取服务端的数据。

缺点：当服务端没有数据时，consumer仍然不断从服务端拉取数据，会消耗consumer一定的资源。

2、push（推）模式

服务端主动推消息给consumer。

优点：服务端只要有数据是才会推数据给consumer。

缺点：服务端发送消息的速率很难适应所有consumer处理消息的速率。

（三）、消费者消费流程

1、总体消费流程

1）消费者通过offset拉取broker中指定位置的消息。offset则保存在系统主题中，系统主题保存在磁盘中，所以即使服务端出现故障或重启等能够按上次消费的位置开始消费。

2）一个消费者可以消费多个分区的数据。

3）每个分区的数据只能由消费者组的一个消费者消费。

image

2、消费者组

1）消费者在消费消息时必须指定消费者组id（group.id），具有相同的group.id组成一个消费者组。

2）消费者组的消费者消费不同分区的数据。一个分区只能由一个消费者消费。

3）消费者组之间相互独立，互不影响。

4）消费者组的消费者数量应小于等于分区数量，若消费者数量大于分区数，多余的消费者则不会消费消息。

5）消费者消费消息后发送offset给服务端。

image

3、消费者初始化流程

image

（四）、分区的分配策略以及再平衡

1、Range分区策略

首先对同一个topic的分区对分区序号进行排序，然后消费者按照字母排序，利用分区数除以消费者数量得出每个消费者平均消费分区数，若剩余多的则按顺序消费。

注意：当topic数量比较多时，使用Range分区策略会导致排在前面的消费总是分配更多的消费者，造成数据倾斜。

再平衡策略：当某个消费者挂了，Kafka会将该消费者的任务全部交给正常消费的消费者。

image

2、RoundRobin分区策略

先进行排序，然后消费依次轮询消费分区数据。

再平衡：触发再平衡后，将故障的消费者的任务轮询交给正常消费的消费者。

image

3、Sticky分区策略

粘性分区是 Kafka 从 0.11.x 版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分****区不变化。**

再平衡*****：****尽可能均衡的随机分成 0 和 1 号分区数据，分别由 1 号消费者或者 2 号消费者消费。*

image

（五）、offset位置

1、offset的维护

Kafka0.9版本之前，consumer默认将offset*****保存在Zookeeper中。*

从0.9版本开始，consumer默认将offset保存在Kafka一个内置的topic中，该topic为__consumer_offsets。

2、offset提交

分为自动提交和手动提交，默认为自动提交

1）自动提交

enable.auto.commit：设置为true

auto.commit.interval.ms：自动提交offset的间隔时间，默认为5s

2）手动提交

　　　　　　　　　enable.auto.commit ：设置为false。用户可以根据具体场景进行提交，更加灵活。手动提交可分为同步提交和异步提交两张方式。

3、指定offset消费

当消费者第一次消费或者offset信息丢失，这是消费者该如何进行消费呢？Kafka提供了三种方式：earliest，latest和none

earliest：自动将offset设置为最开始的位置。

latest：自动将offset设置为最新的offset位置。

none：若没有找到之前的offset信息，则会抛出异常。

4、指定时间消费

消费者根据指定时间（比如1天前）获取数据，通过broker的时间戳索引文件查到该时间对应的offset，然后用该offset获取对应的数据。

（六）、漏消费和重复消费

1、漏消费

先提交offset后处理数据，可能会造成漏消费消息。一般用户设置为手动异步提交offset会引起这种问题。

场景：消费者offset提交方式设置为手动异步提交，消费者把offset提交出去，但是数据还未落盘，此时消费者挂掉，分区已经将最新的offset保存起来了。消费者重启后获取的offset则是最新提交的offset了，造成数据遗漏处理。

2、重复消费

已经消费了消息，但是没提交offset。一般消费者设置为自动提交会有这种问题。

场景：消费者设置offset自动提交时间为5s，在第2s时消费者挂掉，在挂掉前已经处理部分数据，但offset没有提交到broker。消费者重启后拿到的offset则为旧的offset，之前处理过的数据又被拉取一遍，造成重复消费数据。

（七）、消费者事务

若要保证消费者消费数据时不漏消费，不重复消费，则需要将消费者端在消费消息和提交offset的操作看作是依次原子操作。

（八）、数据积压

所谓数据积压是指broker中保存大量未消费的消息。若长时间未消费且触发了删除策略，那么这部分数据就会丢失。可能造成数据积压的原因有消费者的处理能力不足，分区数较少，可适当增加分区数量，使分数量等于消费者数量。也可以把消费者每次拉取数据的大小往上调。

五、总结

Kafka凭借自身优秀的性能和海量数据的处理能力以及数据的可靠性保证，组件在大数据领域占据主流地位。