《零散知识点 · Kafka 知识拓展》

本文主要是介绍《零散知识点 · Kafka 知识拓展》，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

📢 大家好，我是【战神刘玉栋】，有10多年的研发经验，致力于前后端技术栈的知识沉淀和传播。 💗

🌻 CSDN入驻不久，希望大家多多支持，后续会继续提升文章质量，绝不滥竽充数，欢迎多多交流。👍

文章目录

- 写在前面的话
- 环境调优专栏
- 如何防止消息丢失
- 如何防止重复消费
- 如何防止消息乱序
- 如何防止消息积压
- 零散知识收录
- 运维常见问题
- 总结陈词

写在前面的话

本篇文章分享一下关于Kafka实战开发中一些拓展知识点。

环境调优专栏

建议：提升吞吐量

1、提升生产吞吐量

（1）buffer.memory：发送消息的缓冲区大小，默认值是 32m，可以增加到 64m。

（2）batch.size：默认是 16k。如果 batch 设置太小，会导致频繁网络请求，吞吐量下降；如果 batch 太大，会导致一条消息需要等待很久才能被发送出去，增加网络延时。

（3）linger.ms，这个值默认是 0，意思就是消息必须立即被发送。一般设置一个 5-100毫秒。如果 linger.ms 设置的太小，会导致频繁网络请求，吞吐量下降；如果 linger.ms 太长，会导致一条消息需要等待很久才能被发送出去，增加网络延时。

（4）compression.type：默认是 none，不压缩，但是也可以使用 lz4 压缩，效率还是不错的，压缩之后可以减小数据量，提升吞吐量，但是会加大 producer 端的 CPU 开销。

2、增加分区数

（1）如果Kafka集群数量为1（单机版本），则分区数、分区副本数、消费者组内数量，这三个都必须设置1；

（2）如果Kafka集群数量为2（或大于2），则分区数、分区副本数，可以设置2，消费者组内数量，可以设置1-2；

3、消费者提高吞吐量

（1）调整 fetch.max.bytes 大小，默认是 50m。

（2）调整 max.poll.records 大小，默认是 500 条。

4、增加下游消费者处理能力，例如利用异步线程处理方案

建议：数据精确一次

1、生产者角度

⚫ acks 设置为-1 （acks=-1）。

⚫ 幂等性（enable.idempotence = true） + 事务。

2、broker 服务端角度

⚫ 分区副本大于等于 2 （–replication-factor 2）。

⚫ ISR 里应答的最小副本数量大于等于 2 （min.insync.replicas = 2）。

3、消费者角度

⚫ 事务 + 手动提交 offset （enable.auto.commit = false）。

⚫ 消费者输出的目的地必须支持事务（MySQL、Kafka）。

建议：从配置看调优

Kafka在弹性、容错性以及高吞吐量方面有着很大的优势。想要达到生产环境最优，发挥这些特性，需要我们进行一系列的配置。

acks

acks参数指定了必须要有多少个分区副本收到消息，生产者才认为该消息是写入成功的，这个参数对于消息是否丢失起着重要作用，该参数的配置具体如下：

acks=0，表示生产者在成功写入消息之前不会等待任何来自服务器的响应. 换句话说，一旦出现了问题导致服务器没有收到消息，那么生产者就无从得知，消息也就丢失了. 改配置由于不需要等到服务器的响应，所以可以以网络支持的最大速度发送消息，从而达到很高的吞吐量。

acks=1，表示只要集群的leader分区副本接收到了消息，就会向生产者发送一个成功响应的ack，此时生产者接收到ack之后就可以认为该消息是写入成功的. 一旦消息无法写入leader分区副本(比如网络原因、leader节点崩溃),生产者会收到一个错误响应，当生产者接收到该错误响应之后，为了避免数据丢失，会重新发送数据.这种方式的吞吐量取决于使用的是异步发送还是同步发送.

尖叫提示：如果生产者收到了错误响应，即便是重新发消息，还是会有可能出现丢数据的现象. 比如，如果一个没有收到消息的节点成为了新的Leader，消息就会丢失.

acks =all,表示只有所有参与复制的节点(ISR列表的副本)全部收到消息时，生产者才会接收到来自服务器的响应. 这种模式是最高级别的，也是最安全的，可以确保不止一个Broker接收到了消息. 该模式的延迟会很高.

min.insync.replicas

上面提到，当acks=all时，需要所有的副本都同步了才会发送成功响应到生产者. 其实这里面存在一个问题：如果Leader副本是唯一的同步副本时会发生什么呢？此时相当于acks=1.所以是不安全的.

Kafka的Broker端提供了一个参数min.insync.replicas,该参数控制的是消息至少被写入到多少个副本才算是"真正写入",该值默认值为1，生产环境设定为一个大于1的值可以提升消息的持久性. 因为如果同步副本的数量低于该配置值，则生产者会收到错误响应，从而确保消息不丢失.

replica.lag.time.max.ms

In-sync replica(ISR)称之为同步副本，ISR中的副本都是与Leader进行同步的副本，所以不在该列表的follower会被认为与Leader是不同步的. 那么，ISR中存在是什么副本呢？首先可以明确的是：Leader副本总是存在于ISR中. 而follower副本是否在ISR中，取决于该follower副本是否与Leader副本保持了“同步”.

尖叫提示：对于"follower副本是否与Leader副本保持了同步"的理解如下：

(1)上面所说的同步不是指完全的同步，即并不是说一旦follower副本同步滞后与Leader副本，就会被踢出ISR列表.

(2)Kafka的broker端有一个参数replica.lag.time.max.ms, 该参数表示follower副本滞后与Leader副本的最长时间间隔，默认是10秒. 这就意味着，只要follower副本落后于leader副本的时间间隔不超过10秒，就可以认为该follower副本与leader副本是同步的，所以哪怕当前follower副本落后于Leader副本几条消息，只要在10秒之内赶上Leader副本，就不会被踢出出局.

（3）如果follower副本被踢出ISR列表，等到该副本追上了Leader副本的进度，该副本会被再次加入到ISR列表中，所以ISR是一个动态列表，并不是静态不变的。

retries

生产者从服务器收到的错误有可能是临时性的错误（比如分区找不到首领）。在这种情况下， retries参数的值决定了生产者可以重发消息的次数，如果达到这个次数，生产者会放弃重试并返回错误。默认情况下，生产者会在每次重试之间等待100ms ，可以通过retry.backoff.ms 参数来配置时间间隔。

比如，设置了acks=all和min.insync.replicas=2。由于某种原因，所有follower都挂了，由于min.insync.replicas=2，所以生产者无法收到来自Broker端的ack。

此时我们会从Producer端收到一个错误消息：“Broker: Not enough in-sync replicas”。这就意味着Kafka不能在Broker上追加生产的消息(数据)了，因为此时的ISR的数量不够。此时在Broker端会有如下的错误消息：

org.apache.kafka.common.errors.NotEnoughReplicasException: The size of the current ISR Set(0) is insufficient to satisfy the min.isr requirement of 2 for partition

默认情况下，Producer不会对此错误进行处理，这就会造成消息丢失，即**at-most-once **语义。我们可以通过配置重试次数来让生产者重新发送消息。比如配置retries=3，默认为0

enable.idempotence

在某些情况下，实际上已将消息提交给了所有同步副本，但是由于网络问题，Broker无法向Producer发送确认ack。由于我们设置retries=3，所以producer将重新发送消息3次，这可能会导致topic中消息重复。

比如有一个producer向该topic发送1M消息，并且在提交消息之后但在生产者收到所有确认ack之前，broker失败了。在这种情况下，由于重试机制，最终可能在该topic上收到超过1M的消息，这也称为at-lease-once语义。

当然，我们想要实现的是exactly-once语义，即：即便生产者重新发送消息，消费者也应该只收到一次相同的消息。

此时需要进行幂等操作，所谓幂等，即指一次执行一个操作或多次执行一个操作具有相同的效果。配置幂等很简单，通过配置enable.idempotence=true即可,默认为false。

那么，幂等是如何实现的呢？由于消息是分batch(批次)发送的，每个batch都有一个序列号。在Broker端，会追踪每个分区的最大序列号。如果出现序列号较小或相等的batch(批次)，broker将不会将该batch写入topic。这样，除了保证了幂等性，还可以确保batch的顺序。

max.in.flight.requests.per.connection

该参数指定了生产者在收到服务器晌应之前可以发送多少个消息。它的值越高，就会占用越多的内存，不过也会提升吞吐量。把它设为1可以保证消息是按照发送的顺序写入服务器的，即使发生了重试。

因为如果将两个批次发送到单个分区，并且第一个批次失败并被重试，但是，接着第二个批次写入成功，则第二个批次中的记录可能会首先出现，这样就会发生乱序。

如果没有启用幂等功能，但仍然希望按顺序发送消息，则应将此设置配置为1。但是，如果已经启用了幂等，则无需显式定义此配置。

buffer.memory

该参数用来设置生产者内存缓冲区的大小，生产者用它缓冲要发送到服务器的消息。如果应用程序发送消息的速度超过发送到服务器的速度，会导致生产者空间不足。这个时候，send()方法调用要么被阻塞，要么抛出异常，取决于如何设置max.block.ms。

当生产者调用时send()，消息并不会立即发送，而是会添加到内部缓冲区中。默认buffer.memory值为32MB。如果生产者发送消息的速度超过了将消息发送到broker的速度，或者存在网络问题，send()方法调用会被阻塞max.block.ms参数配置的时常，默认1分钟。

max.block.ms

该参数指定了在调用send()方法或使用partitionsFor()方法获取元数据时生产者的阻塞时间。当生产者的发送缓冲区已满，或者没有可用的元数据时，这些方法就会被阻塞。在阻塞时间达到max.block.ms时，生产者会抛出超时异常。

linger.ms

该参数指定了生产者在发送批次之前等待更多消息加入批次的时间。kafka生产者会在批次填满或linger.ms达到上限时把批次发送出去。默认情况下，只要有可用的线程，生产者就会把消息发送出去，就算批次里只有一个消息。把linger.ms设置成比0大的数，让生产者在发送批次之前等待一会儿，使更多的消息加入到这个批次。虽然这样会增加延迟，但也会提升吞吐量（因为一次性发送更多的消息，每个消息的开销就变小了）。

batch.size

当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算（而不是消息个数）。当批次被填满，批次里的所有消息会被发送出去。不过生产者井不一定都会等到批次被填满才发送，这取决于linger.ms的配置，比如如果linger.ms时间到了，即便批次只包含一个消息，也会被立即发送。所以就算把批次大小设置得很大，也不会造成延迟，只是会占用更多的内存而已。但如果设置得太小，因为生产者需要更频繁地发送消息，会增加一些额外的开销。

可以使用配置使用linger.ms和batch.size。linger.ms是准备好发送批次之前的延迟时间，默认值为0。这意味着即使批次中只有1条消息，批次也会立即发送。有时，会增加linger.ms以减少请求数量并提高吞吐量。但这将导致更多消息保留在内存中。batch.size是单个批次的最大大小，当满足这两个要求中的任何一个时，将发送批次。

compression.type

默认情况下，消息发送时不会被压缩。该参数可以设置为snappy 、gzip 或lz4 ，它指定了消息被发送给broker 之前使用哪一种压缩算也进行压缩。使用压缩可以降低网络传输开销和存储开销，而这往往是向Kafka 发送消息的瓶颈所在。

如何防止消息丢失

总结陈词：关于消费者自动提交到底会不会更容易引发，出现多个版本，但目前个人认为，最新版本的Kafka自动提交逻辑应该是poll的时候是提交上一次的偏移量，而不是当前拉到最新的

消息丢失观点分析一

消息队列发送消息和消费消息的过程，共分为三段，生产过程、服务端持久化过程、消费过程，如下图所示。

这三个过程都有可能弄丢消息，具体如下：

1、生产者解决方案：

1）使用同步发送

非必要，主要为了明确消息发送是否失败，异步回调有处理也可以

发送的时候，只要出现异常，其实不算丢失消息，因为消息都没发成功

另外，捕捉异常再发一次，或错误记录存表，手动重试

2）把ack设成-1或者all，并且设置同步的分区数>=2

前面生产者专栏总结过：

至少一次（（At Least Once ）=ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

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

2、Broker：配置手动刷盘，例如：

设置每1000条消息刷一次盘
flush.messages = 1000

设置每秒刷一次盘
flush.ms = 1000

3、消费端：

Kafka有偏移量的概念，consumer从partition中拉取消息，consumer本地处理完成后需要commit一下offset，表示消费完成，下次就不会再拉取到这条消息。

所以，我们需要关闭自动commit offset的配置，防止consumer拉到消息后，服务宕机，导致消息丢失。

enable.auto.commit = false

PS：如果采用自动提交，如果内部逻辑都是同步的，产生异常也不一定会产生消息丢失。

消息丢失观点分析2

1、消息生产者：

可靠性主要和ack参数有关，ack的三种情况：

● 0：生产者发送过来的数据，不需要等数据落盘应答。

表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消息。大数据统计报表场景，对性能要求很高，对数据丢失不敏感的情况可以用这种。

● 1：生产者发送过来的数据，Leader收到数据后应答。

至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。

● -1 (all) ：生产者发送过来的数据，Leader+和isr队列里面的所有节点收齐数据后应答。

这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。当然如果min.insync.replicas 配置的是1则也可能丢消息，跟acks=1情况类似。

2、消息消费者：

如果消费这边配置的是自动提交，万一消费到数据还没处理完，就自动提交offset了，但是此时你consumer直接宕机了，未处理完的数据丢失了，下次也消费不到了。

可以采用手动提交方式，由程序负责控制。

消息丢失观点分析3

观点1：包括尚硅谷在内，以及部分帖子，认为自动提交反而不会引发漏消费，如下图。

观点2：部分帖子认为“自动提交会丢消息，因为消费者在消费前提交offset，有可能提交完后还没消费时消费者挂了。

如何防止重复消费

消息生产者：

发送消息如果配置了重试机制，比如网络抖动时间过长导致发送端发送超时，实际broker可能已经接收到消息，但发送方会重新发送消息。

生产者数据重复分为如下情况：

• 至少一次（（At Least Once ）= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

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

• 最多一次（（At Most Once ）= ACK级别设置为0

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

• 精确一次（（Exactly Once））：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。

要保证消息是精确一次，做法就是：幂等性 + 至少一次

幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。

消息消费端：

如果消费这边配置的是自动提交，刚拉取了一批数据处理了一部分，但还没来得及提交，服务挂了，下次重启又会拉取相同的一批数据重复处理。

因此，一般消费端都是要做消费幂等处理的。

所谓的幂等性，就是多次访问的结果是⼀样的。

对于 Rest 的请求 get（查询、幂等）、post（新增、非幂等）、put（修改、幂等）、delete（删除、幂等）。

即需要对Post 操作做非幂等处理，具体方案，可以将业务ID设置为主键，或者使用Redis或zk的分布式锁。

实际分析：

HIS5.0的消息中心，自身在生产者和消费者逻辑，也都做了msgNo的消息查重判断。

对于生产者而言多了一重保障，重点需要考虑消息丢失问题。

对于消费者而言，实现了消费幂等处理。

如何防止消息乱序

PS：kafka的顺序消费使用场景不多，因为牺牲掉了性能。

问题描述：

如果发送端配置了重试机制，kafka不会等之前那条消息完全发送成功才去发送下一条消息，这样可能会出现：发送了1，2，3条消息，第一条超时了，后面两条发送成功，再重试发送第1条消息，这时消息在broker端的顺序就是2，3，1了。

所以，是否一定要配置重试要根据业务情况而定。也可以用同步发送的模式去发消息，当然acks不能设置为0，这样也能保证消息发送的有序。

kafka 保证全链路消息顺序消费，需要从发送端开始，将所有有序消息发送到同一个分区，然后用一个消费者去消费，但是这种性能比较低，可以在消费者端接收到消息后将需要保证顺序消费的几条消费发到内存队列(可以搞多个)，一个内存队列开启一个线程顺序处理消息。

如何做到顺序消费

发送方：在发送时将ack不能设置0 ，关闭重试，使用同步发送，等到发送成功再发送下一条，确保消息是顺序发送的。

接收方：消息是发送到一个分区中，只能有一个消费组的消费者来接收消息，因此，kafka的顺序消费会牺牲掉性能。

实际分析：

HIS5.0的消息中心，其运用场景并不注重顺序，因此不用特别考虑。

补充：Kafka 中是怎么体现消息顺序性的？

在每个分区内，每条消息都有offset，所以消息在同一分区内有序，但无法做到全局的有序性。

如何防止消息积压

问题背景

消息的消费者的消费速度远赶不上生产者的生产消息的速度，导致kafka中有大量的数据没有被消费。随着没有被消费的数据堆积越多，消费者寻址的性能会越来越差，最后导致整个kafka对外提供的服务的性能很差，从而造成其他服务也访问速度变慢，造成服务雪崩。

解决方案

1、在一个消费者中启动多个线程，让多个线程同时消费，提升一个消费者的消费能力；

2、如果方案一还不够的话，这个时候可以启动多个消费者，多个消费者部署在不同的服务器上；

零散知识收录

Kafka 中的分区器、序列化器、拦截器是否了解？它们之间的处理顺序是什么？

分区器Partitioner用来对分区进行处理的，即消息发送到哪一个分区的问题。序列化器，这个是对数据进行序列化和反序列化的工具。拦截器，即对于消息发送进行一个提前处理和收尾处理的类Interceptor。

处理顺利首先通过拦截器=>序列化器=>分区器

Kafka 生产者客户端的整体结构是什么样子的？使用了几个线程来处理？分别是什么？

使用两个线程：main 和 sender 线程，main线程会一次经过拦截器、序列化器、分区器将数据发送到 RecoreAccumulator线程共享变量，再由 sender 线程从共享变量中拉取数据发送到 kafka 集群。

参考下面图示，详情参考上述生产者专栏。

消费组中的消费者个数如果超过 topic 的分区，那么就会有消费者消费不到数据

这句话是对的，超过分区个数的消费者不会在接收数据，主要原因是一个分区的消息只能够被一个消费者组中的一个消费者消费。

详情参考分区和消费者数的对应关系。

消费者提交消费位移时提交的是当前消费到的最新消息的 offset 还是 offset+1？

生产者发送数据的offset是从0开始的，消费者消费的数据的offset是从1开始，故最新消息是offset+1

有哪些情形会造成重复消费？

先消费后提交offset，如果消费完宕机了，则会造成重复消费

那些情景会造成消息漏消费？

先提交offset，还没消费就宕机了，则会造成漏消费

当你使用 kafka-topics.sh 创建（删除）了一个 topic 之后， Kafka 背后会执行什么逻辑？

会在 zookeeper 中的/brokers/topics 节点下创建一个新的 topic 节点，如：/brokers/topics/first

触发 Controller 的监听程序

kafka Controller 负责 topic 的创建工作，并更新 metadata cache

topic 的分区数可不可以增加？如果可以怎么增加？如果不可以，那又是为什么？

可以增加，修改分区个数–alter可以修改分区个数

topic 的分区数可不可以减少？如果可以怎么减少？如果不可以，那又是为什么？

不可以减少，减少了分区之后，之前的分区中的数据不好处理

Kafka 有内部的 topic 吗？如果有是什么？有什么所用？

有，__consumer_offsets主要用来在0.9版本以后保存消费者消费的offset

Kafka 分区分配的概念？

Kafka分区对于Kafka集群来说，分区可以做到负载均衡，对于消费者来说分区可以提高并发度，提高读取效率

简述 Kafka 的日志目录结构？

每一个分区对应着一个文件夹，命名为topic-0/topic-1…，每个文件夹内有.index和.log文件。

如果我指定了一个 offset， Kafka Controller 怎么查找到对应的消息？

offset表示当前消息的编号，首先可以通过二分法定位当前消息属于哪个.index文件中，随后采用seek定位的方法查找到当前offset在.index中的位置，此时可以拿到初始的偏移量。通过初始的偏移量再通过seek定位到.log中的消息即可找到。

聊一聊 Kafka Controller 的作用？

Kafka集群中有一个broker会被选举为Controller，负责管理集群broker的上下线、所有topic的分区副本分配和leader的选举等工作。Controller的工作管理是依赖于zookeeper的。

Kafka 中有那些地方需要选举？这些地方的选举策略又有哪些？

在ISR中需要选举出Leader，选择策略为先到先得。在分区中需要选举，需要选举出Leader和follower。

失效副本是指什么？有那些应对措施？

失效副本为速率比leader相差大于10s的follower，ISR会将这些失效的follower踢出，等速率接近leader的10s内，会重新加入ISR

Kafka 的哪些设计让它有如此高的性能？

Kafka天生的分布式架构

对log文件进行了分segment，并对segment建立了索引

对于单节点使用了顺序读写，顺序读写是指的文件的顺序追加，减少了磁盘寻址的开销，相比随机写速度提升很多

使用了零拷贝技术，不需要切换到用户态，在内核态即可完成读写操作，且数据的拷贝次数也更少。

Topic 何时创建？

如果 broker 端配置参数 auto.create.topics.enable 设置为true（默认值就是true），那么当生产者向一个尚未创建的主题发送消息时，会自动创建一个分区数为num.partitions （默认值为1）、副本因子为default.replication.factor（默认值为1）的主题。

Partition 分区机制

分区是kafka中逻辑上的一个概念，一个主题可以保存在多个分区，一个分区只属于一个主题。

不同分区下的消息是不同的，也就是主题的消息是分布在不同分区的。当我们发送一条消息到某个topic时，Kafka 会根据一定的规则来计算出这个消息要保存在哪个分区。有了分区的概念，kafka就可以将不同分区保存在不同服务器，这样可以减轻单个服务器的压力。topic和分区的关系可以通过下图来理解。从生产者角度来看，如果对分区没有特别要求，只需要指定将消息发到哪个topic即可，kafka会自动将消息均衡到不同分区。

消费者组机制

Kafka创建消费者时，必须指定消费者组，对于topic中的某个消息，会发向所有消费者组，但是只会选择消费者组中一个消费者进行发送。也就是说不同消费者组之间的消费是互不影响的，同一个消费者组中的消费者分摊消费消息。

消息是按topic的分区分摊给同一个组中的消费者的，比如一个topic有4个分区，一个消费者组中有两个消费者，那么每个消费者就负责消费某2个分区中的消息。消费者组，消费者以及分区的关系可以通过下图理解。

消费者停机期间发送的消息，消费者启动后还能否消费到？

我们通过实际测试来看下，首先我新建了一个分区数为1的topic，名为demo2，然后先不启动消费者，我向demo2发送了两条消息hello111和hello222，此时再启动消费者，然后再发送一条hello333的消息，下面是运行结果，可以看到，消费者只消费到了偏移为2的hello333这一条消息，也就是没有消费到未启动时发送的消息。

此时我们再关闭消息者，然后向demo2发送一条消息hello444，然后再启动消费者，可以看到消费者消费到了消息。

也就是说，如果我们一个消费者组还从来没从一个topic中消费过消息，那么刚启动时是消费不到历史消息的，如果消费过，那么中间消费者停止，然后后来又启动，消费者停止期间发送的消息再启动后是可以消费到的。

kafka中每条消息都是有个编号的，每个分区的消息按收到的顺序依次递增编号，我们在收到的消息中打印的offset就是这个消息的偏移量，也就是这个消息是分区中的第几条消息。kafka会记录下某个消费者组消费到了哪条消息，称为这个消费者组的消费位移，下次消费者启动时就可以从记录的消费位移开始消费，而当一个新的消费者组消费时，kafka中没有消费位移的信息，所以也就没法消费历史消息。

上面提到的情况是我们按照默认配置时的情况，其实kafka提供了一个配置参数auto.offset.reset来控制如何进行消费。auto.offset.reset的取值有3个，分别为latest、earliest、none，默认值为latest，也就是当一个消费者找不到消费位移信息时，会从最新的消息（也就是消费者启动后发送的消息）开始消费，earliest就是从最早的消息开始消费，这种情况下就可以消费到在消费者启动之前发送的消息。none表示当找不到消费位移信息时，kafka会报错。

kafka还可以指定从任意位置的消息开始消费，对应Java客户端的consumer.seek()方法。

Kafka 分布式

为什么说kafka是分布式模型呢？首先，同一个kafka集群有共同拥有一个topic, 而同一个topic又拥有不同的分区，不同的分区可以分布在不同的borker上也就是不同的机子上，所以，分区是分布式的，则数据也是分布式的，kafka就是分布式

运维常见问题

Q：外网 Kafka的配置，Java如何访问？

SB 连接Kafka 提示错误：

Connection to node 1 (localhost/127.0.0.1:9092) could not be established.

检查阿里云安全组配置，修改Kafka配置文件，如下：

listeners=PLAINTEXT://0.0.0.0:9092

advertised.listeners=PLAINTEXT://47.95.248.102:9092

Q：运行 zookeeper，出现报错 Classpath is empty ？

A：kafka 安装路径中有空格，解决空格问题即可，例如使用 mv 语法。但如果路径中没有空格呢？可能是你下载的kafka有问题，不能下载 src 的版本。

Q：操作 Kafka 提示 zookeeper is not a recognized option

操作命令：bin/kafka-topics.sh --list --zookeeper localhost:2181

查阅资料后发现是kafka的版本问题，低版本的kafka可以使用以上的命令，但是在高版本的kafka中需要使用如下命令才行：bin/kafka-topics.sh --list --bootstrap-server localhost:9092

Q：报错：: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partition.

报错原因：producer在向kafka broker写的时候，刚好发生选举，本来是向broker0上写的，选举之后broker1成为leader，所以无法写成功，就抛异常了。

解决办法：修改producer的重试参数retries参数，默认是0，一般设置为3，我在生产环境配置的retries=10

Q：java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for binlogCsbbroker-2 due to 30026 ms has passed since batch creation plus linger time

报错原因：具体原因我自己还没有找到，但是网友们都说是因为kafka在批量写的时候，这一批次的数据没有在30s内还处理完，（30s为request.timeout.ms默认值），这一批次的数据就过期了，所以抛出异常

解决办法：增大request.timeout.ms，我在生产环境配置的是request.timeout.ms=60000 // 由原来默认的30s改成60s

Q：java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.NetworkException: The server disconnected before a response was received.

报错原因：kafka client与broker断开连接了

解决办法：重启服务

**Q：消费者程序消费Kafka异常，2023-03-24 00:54:54.061 WARN [msg-consumer-service,] 8 — [ad | producer-1] org.apache.kafka.clients.NetworkClient : [Producer clientId=producer-1] Error while fetching metadata with correlation id 19109 : {WXNoticeAppointment=UNKNOWN_SERVER_ERROR} **

解决办法： kafka先把副本置0，重启zk后再把kafka副本置1，因为这两个有依赖关系，zk没做持久化，重启就算删数据，kafka要重启就是这个步骤。