1.首先一个kafka集群有很多个kafka的服务器，每个kafka服务器就是一个broker，每一类消息有一个topic，生产者将一个消息发送给broker。

2.每个topic会有一个或者多个分区，broker根据分发机制将这个消息分给这个topic下的某个分区的leader，

分发机制：

* 1.发的消息指定了分区就发到特定分区下

* 2.指定了key，就根据murmur2 哈希算法对key计算得到一个哈希值，将哈希值与分区数量取余，得到分区。

* 3.没有指定分区，也没有指定key，那么就根据一个自增计数与分区数取余得到分区，这样可以让消息分发在每个分区更加均匀。

3.每个分区就是一个目录，目录名是topic+分区编号，在收到消息后会将消息写入到日志文件中，如果一个分区的消息都有存放在一个日志文件中，那么文件会比较大，查询时会比较慢，而且也不便于之后删除旧的消息。所以每个分区对应多个大小相等的segment文件，每个segment的名称是上一个segment最后一条消息的offset，一个segment有两个文件，一个是.index文件，记录了消息的offset及这条消息数据在log文件中的偏移量。一个是.log文件，实际存储每个消息数据，每条消息数据大小不一，每条消息数据包含offset，消息体大小，消息体等等内容。查的时候根据offset先去index文件找到偏移量，然后去log文件中读。

(具体的segment切分有很多个触发条件：

当log文件>log.segment.bytes时切分，默认是1G。

或者是segment文件中最早的消息距离现在的时间>log.roll.ms配置的时间，默认是7天。

或者是索引文件index>log.index.size.max.bytes的大小，默认是10M。）

4.分区leader将消息存储到日志文件中后还不能算是写成功，会把消息同步给所有follower，当follower同步好消息之后就会给leader发ack，leader收到所有follower返回的ack之后，这条才算是写成功，然后才会给生产者返回写成功。（依据ACK配置来决定多少follower同步成功才算生产者发送消息成功）

5.消费者读数据时就去分区的leader中去读，一个消费者可以消费多个分区，但是一个分区只能一个消费者来消费，默认消费者取完数据就会自动提交，一般会关闭自动提交，消费者消费成功后，进行手动提交，分区的offset才会向后移动。（默认是会自动提交，一般会关闭自动提交）

##### 注意事项：

1.replication.factor>=2，也就是一个分区至少会有两个副本。

2.min.insync.replicas默认是1，leader至少要有一个follow跟自己保持联系没有掉线。(这个配置只有在ack为all或者-1时有用，也就是ack为all也只是要求生产者发送的消息，被leader以及ISR集合里面的从节点接收到，就算所有节点都接收到了。)

3.一般设置了ack=all就不会丢数据。因为会保证所有的follower都收到消息，才算broker接收成功，默认ack=1。

4.retries=，生产者写入消息失败后的重试次数。

5.每个partition有一个offset，

6.生产者ACK配置：

**1（默认）** 数据发送到Kafka后，经过leader成功接收消息的的确认，就算是发送成功了。在这种情况下，如果leader宕机了，则会丢失数据。

**0** 生产者将数据发送出去就不管了，不去等待任何返回。这种情况下数据传输效率最高，但是数据可靠性确是最低的。

**-1** 也就是all，producer需要等待ISR中的所有follower都确认接收到数据后才算一次发送完成，可靠性最高。

这块比较常见的一个场景，就是 `Kafka` 某个 `broker` 宕机，然后重新选举 `partition` 的 `leader` 。大家想想，要是此时其他的 `follower` 刚好还有些数据没有同步，结果此时 `leader` 挂了，然后选举某个 `follower` 成 `leader` 之后，不就少了一些数据？这就丢了一些数据啊。

此时一般是要求起码设置如下 4 个参数：

- 给 `topic` 设置 `replication.factor` 参数：这个值必须大于 1，要求每个 `partition` 必须有 **至少** 2 个副本。

- 在 `Kafka` 服务端设置 `min.insync.replicas` 参数：这个值必须大于 1，这个是 **要求一个 leader 至少感知到有至少一个 follower 还跟自己保持联系**，没掉队，这样才能确保 `leader` 挂了还有一个 `follower` 吧。

- 在 `producer` 端设置 `acks=all`：这个是要求每条数据，**必须是写入所有 replica 之后，才能认为是写成功了**。

- 在 `producer` 端设置 `retries=MAX`（很大很大很大的一个值，无限次重试的意思）：这个是要求**一旦写入失败，就无限重试**，卡在这里了。

这样配置之后，至少在Kafka `broker` 端就可以保证在`leader` 所在 `broker` 发生故障，进行`leader` 切换时，数据不会丢失。

如果按照上述的思路设置了`acks=all`，一定不会丢，要求是，你的 `leader` 接收到消息，所有的`follower` 都同步到了消息之后，才认为本次写成功了。如果没满足这个条件，生产者可以自动不断的重试，重试无限次。

#### 怎么实现 Exactly-Once？

##### 生产端幂等性发送

为了实现Producer的幂等语义，Kafka引入了`Producer ID`（即`PID`）和`Sequence Number`。每个新的Producer在初始化的时候会被分配一个唯一的PID，该PID对用户完全透明而不会暴露给用户。

对于每个PID，该Producer发送数据的每个`<Topic, Partition>`都对应一个从0开始单调递增的`Sequence Number`。

类似地，Broker端也会为每个`<PID, Topic, Partition>`维护一个序号，并且每次Commit一条消息时将其对应序号递增。对于接收的每条消息，如果其序号比Broker维护的序号（即最后一次Commit的消息的序号）大一，则Broker会接受它，否则将其丢弃：

- 如果消息序号比Broker维护的序号大一以上，说明中间有数据尚未写入，也即乱序，此时Broker拒绝该消息，Producer抛出`InvalidSequenceNumber`

- 如果消息序号小于等于Broker维护的序号，说明该消息已被保存，即为重复消息，Broker直接丢弃该消息，Producer抛出`DuplicateSequenceNumber`

上述设计解决了0.11.0.0之前版本中的两个问题：

- Broker保存消息后，发送ACK前宕机，Producer认为消息未发送成功并重试，造成数据重复

- 前一条消息发送失败，后一条消息发送成功，前一条消息重试后成功，造成数据乱序。

http://www.jasongj.com/kafka/transaction/

##### 消费端幂等性

只能自己从业务层面保证重复消费的幂等性，例如引入版本号机制。

#### 事务性保证

上述幂等设计只能保证单个Producer对于同一个`<Topic, Partition>`的`Exactly Once`语义。

另外，它并不能保证写操作的原子性——即多个写操作，要么全部被Commit要么全部不被Commit。

更不能保证多个读写操作的的原子性。尤其对于Kafka Stream应用而言，典型的操作即是从某个Topic消费数据，经过一系列转换后写回另一个Topic，保证从源Topic的读取与向目标Topic的写入的原子性有助于从故障中恢复。

事务保证可使得应用程序将生产数据和消费数据当作一个原子单元来处理，要么全部成功，要么全部失败，即使该生产或消费跨多个`<Topic, Partition>`。

另外，有状态的应用也可以保证重启后从断点处继续处理，也即事务恢复。

为了实现这种效果，应用程序必须提供一个稳定的（重启后不变）唯一的ID，也即`Transaction ID`。`Transactin ID`与`PID`可能一一对应。区别在于`Transaction ID`由用户提供，而`PID`是内部的实现对用户透明。

另外，为了保证新的Producer启动后，旧的具有相同`Transaction ID`的Producer即失效，每次Producer通过`Transaction ID`拿到PID的同时，还会获取一个单调递增的epoch。由于旧的Producer的epoch比新Producer的epoch小，Kafka可以很容易识别出该Producer是老的Producer并拒绝其请求。

有了`Transaction ID`后，Kafka可保证：

- 跨Session的数据幂等发送。当具有相同`Transaction ID`的新的Producer实例被创建且工作时，旧的且拥有相同`Transaction ID`的Producer将不再工作。

- 跨Session的事务恢复。如果某个应用实例宕机，新的实例可以保证任何未完成的旧的事务要么Commit要么Abort，使得新实例从一个正常状态开始工作。

需要注意的是，上述的事务保证是从Producer的角度去考虑的。从Consumer的角度来看，该保证会相对弱一些。尤其是不能保证所有被某事务Commit过的所有消息都被一起消费，因为：

- 对于压缩的Topic而言，同一事务的某些消息可能被其它版本覆盖

- 事务包含的消息可能分布在多个Segment中（即使在同一个Partition内），当老的Segment被删除时，该事务的部分数据可能会丢失

- Consumer在一个事务内可能通过seek方法访问任意Offset的消息，从而可能丢失部分消息

- Consumer可能并不需要消费某一事务内的所有Partition，因此它将永远不会读取组成该事务的所有消息

1. **异步通信**：有些业务不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

2. **解耦**：降低工程间的强依赖程度，针对异构系统进行适配。在项目启动之初来预测将来项目会碰到什么需求，是极其困难的。通过消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口，当应用发生变化时，可以独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束

3. **冗余**：有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

4. **扩展性**：因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。便于分布式扩容

5. **过载保护**：在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量无法提取预知；如果以为了能处理这类瞬间峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃

6. **可恢复性**：系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

7. **顺序保证**：在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。

8. **缓冲**：在任何重要的系统中，都会有需要不同的处理时间的元素。消息队列通过一个缓冲层来帮助任务最高效率的执行，该缓冲有助于控制和优化数据流经过系统的速度。以调节系统响应时间。

9. **数据流处理**：分布式系统产生的海量数据流，如：业务日志、监控数据、用户行为等，针对这些数据流进行实时或批量采集汇总，然后进行大数据分析是当前互联网的必备技术，通过消息队列完成此类数据收集是最好的选择

##### MQ缺点

1. 系统可用性降低：系统引入的外部依赖越多，越容易挂掉。本来你就是 `A` 系统调用 `BCD` 三个系统的接口就好了， `ABCD` 四个系统好好的，没啥问题，你偏加个 `MQ` 进来，万一 `MQ` `挂了咋整，MQ` 一挂，整套系统崩溃的，你不就完了？如何保证消息队列的高可用。

2. 系统复杂度提高：硬生生加个 `MQ` 进来，你怎么保证消息没有重复消费？怎么处理消息丢失的情况？怎么保证消息传递的顺序性？头大头大，问题一大堆，痛苦不已。

3. 一致性问题： `A` 系统处理完了直接返回成功了，人都以为你这个请求就成功了；但是问题是，要是 `BCD` 三个系统那里， `BD` 两个系统写库成功了，结果 `C` 系统写库失败了，咋整？你这数据就不一致了。

#### ISR是什么？

ISR（in-sync replica） 就是 Kafka 为某个分区维护的一组同步集合，即每个分区都有自己的一个 ISR 集合，就是从分区的从节点中找出一些节点加入到ISR集合（min.insync.replicas这个参数设定ISR中的最小副本数是多少，默认值为1）。处于 ISR 集合中的副本，意味着 follower 副本与 leader 副本保持同步状态，只有处于 ISR 集合中的副本才有资格被选举为 leader。follower从leader同步数据有一些延迟（延迟时间replica.lag.time.max.ms），一旦超过延迟时间，就会把这个这个follower从ISR列表中移除。被移除的followe会从leader复制数据进行追赶，一旦追赶上又可以重新进入ISR列表。一条 Kafka 消息，只有被 ISR 中的副本都接收到，才被视为“已同步”状态。这跟 zk 的同步机制不一样，zk 只需要超过半数节点写入，就可被视为已写入成功。

传统的IO接口像read和write系统调用，在执行过程中都是涉及到数据拷贝操作的，比如调用read()接口去读取一个文件时，首先需要将CPU由用户切换成内核态，然后把文件从磁盘读取到

#### read()和write()


**read()系统调用的步骤：**

1.会涉及到到一次用户态到内核态的切换，然后会发出 sys_read()系统调用，从文件读取数据。(一次上下文切换)

2.磁盘控制器会使用DMA技术将磁盘文件拷贝到内核内存空间的缓冲区。(一次DMA拷贝)

3.CPU会将数据从内核内存空间的缓冲区拷贝到用户进程内存空间的缓冲区。(一次CPU拷贝)

4.然后read()系统调用返回后，会进行内核态往用户态的切换，这样用户程序进程就可以修改数据了。(一次上下文切换)

**write()系统调用的步骤：**

1.首先会涉及CPU从用户态切换到内核态，然后会将数据从用户程序的内存空间拷贝到内核内存空间中的Socket缓冲区。(一次上下文切换，一次CPU拷贝)

2.网卡会使用DMA技术，将数据从内核内存空间中的缓冲区拷贝到网卡。(一次DMA拷贝)

3.write()调用完成后会从内核态切换到用户态。(一次上下文切换)

#### 2.MMAP和write()


##### mmap

1.CPU从用户态切换到内核态，磁盘控制器使用DMA技术将数据从磁盘拷贝到内核的内存空间。不会将数据拷贝到用户程序的内存空间，而是将一块物理内存让用户进程的空间与内核空间进行共享，将内核中的这部分内存空间映射到用户进程的内存空间，从而让用户进程可以直接访问这部分内存。(一次上下文切换，一次DMA拷贝)

2.mmap调用完毕后，CPU会从内核态切换到用户态。(一次上下文切换)

mmap相比于read()系统调用还是会有2次上下文切换，但是可以减少一次CPU拷贝，因为数据是存在内核的内存空间中。

##### write

1.首先CPU从用户态切换到内核态，然后把数据从内核的内存空间拷贝到内核中Socket缓冲区。(一次上下文切换，一次CPU拷贝)

2.网卡使用DMA技术，将数据从Socket缓冲区拷贝到网卡。发送完毕后，从内核态切换为用户态。(一次上下文切换，一次DMA拷贝)

https://mp.weixin.qq.com/s/xDZ9NnyUZSoR9npuMLdpWA

https://blog.csdn.net/choumu8867/article/details/100658332

#### sendfile

这种方式只能用于发送文件，不能修改文件，在Kakfa发送消息给消费者时有用到。


读取时：

1.首先CPU从用户态切换成内核态，然后磁盘控制器使用DMA技术将文件从磁盘拷贝到内核空间的缓冲区中。

（一次上下文切换，一次DMA拷贝）

发送时：

2.早期的版本是将数据从内核空间中的缓存区拷贝到内核空间的Socket缓冲区，在Linux 2.4以后，是只需要将数据在内核空间的文件数据缓存中的位置和偏移量写入到Socket缓存中，然后网卡直接从Socket缓存中读取文件的位置和偏移量，使用DMA技术拷贝到网卡。发送完毕后，从内核态切换为用户态。

（一次上下文切换，一次DMA拷贝。）

##### 总结：

传统read()和write()方案：数据拷贝了4次，CPU上下文切换了很多次

mmap和write()方案：数据拷贝了3次，会减少一次CPU拷贝，上下文切换了4次。(可以减少1次CPU拷贝)

sendfile方案：数据拷贝了2次，上下文切换了2次。但是用户进程不能修改数据。(可以减少2次CPU拷贝，至少2次上下文切换)

log.flush.interval.messages 最大刷盘消息数量

log.flush.interval.interval.ms 最大刷盘时间间隔

log.flush.scheduler.interval.ms 定期刷盘间隔

可以通过设置 最大刷盘消息数量 和 最大刷盘时间间隔 来控制fsync系统调用的时间，但是Kafka不推荐去设置这些参数，希望让操作系统来决定刷盘的时机，这样可以支持更高的吞吐量。而且Kafka保证可用性是通过多副本来实现的，一个机器挂掉了就会选举副本作为leader。

1.topic下分区的数量增加了或者减少了。(这个一般是我们手动触发的)

2.消费者的数量发生了改变，例如新增加了消费者或者有消费者挂掉了。

Kafka有一个session.timeout.ms，最大会话超时时间，最长是10s。就是如果broker与消费者之间的心跳包超过10s还没有收到回应，就会认为消费者掉线了。以及还有一个max.poll.interval.ms参数，消费者两次去broker拉取消息的间隔，默认是5分钟。如果消费者两次拉取消息的间隔超过了5分钟，就会认为消费者掉线了。

一旦发生rebalance了，有可能会导致重复消费的问题，就是消费者A拉取了100条消息，消费时间超过了5分钟，被broker认定下线，就会进行rebalance，把这个分区分配给其他消费者消费，其他消费者就会进行重复消费。

怎么解决rebalance带来的重复消费问题呢？

1.可以减少每批消息的处理时间，让每条消息的处理时间减少，或者是修改max.poll.records，减小每次拉取消息的数量。

2.可以自行在MySQL或者Redis里面存储每个分区消费的offset，然后消费者去一个新的分区拉取消息时先去读取上次消费的offset。

3.为消息分配一个唯一的消息id，通过消息id来判定是否重复消费了。

##### kafka 1.1的优化

新版本新增了**group.initial.rebalance.delay.ms**参数。空消费组接受到成员加入请求时，不立即转化到PreparingRebalance状态来开启reblance。当时间超过**group.initial.rebalance.delay.ms**后，再把group状态改为PreparingRebalance（开启reblance），这样可以避免服务启动时，consumer陆续加入引起的频繁Rebalance。

##### Kafka2.3对reblance的优化

但对于运行过程中，consumer超时或重启引起的reblance则无法避免，其中一个原因就是，consumer重启后，它的身份标识会变。简单说就是Kafka不确认新加入的consumer是否是之前挂掉的那个。

在Kafka2.0中引入了静态成员ID，使得consumer重新加入时，可以保持旧的标识，这样Kafka就知道之前挂掉的consumer又恢复了，从而不需要Reblance。这样做的好处有两个：

1. 降低了Kafka Reblance的频率

2. 即使发生Reblance，Kafka尽量让其他consumer保持原有的partition，减少了重分配引来的耗时、幂等等问题

https://blog.csdn.net/weixin_37968613/article/details/104607012

https://blog.csdn.net/z69183787/article/details/105138782

https://zhuanlan.zhihu.com/p/87577979

https://www.cnblogs.com/runnerjack/p/12108132.html

https://blog.csdn.net/qq_37142346/article/details/91349100

https://honeypps.com/mq/kafka-basic-knowledge-of-selection/