深入理解Kafka核心设计与

Topic参数/Broker参数

cleanup.policy/log.cleanup.policy：日志压缩策略。默认delete，可配compact。

compression.type/compression.type：消息的压缩类型。默认producer，保留原始压缩类型。可配uncompressed、snappy、lz4、gzip。

delete.retention.ms/log.cleaner.delete.retention.ms：被标识为删除的数据能保存多久。默认1天。

file.delete.delay.ms/log.segment.delete.delay.ms：清理文件之前可以等待多长时间，默认1分钟。

flush.messages/log.flush.interval.messages：需要收集多少消息才能强制刷新到磁盘，默认Long.MAX_VALUE。让OS决定。

flush.ms/log.flush.interval.ms：需要等待多久将消息强制刷新到磁盘，默认Long.MAX_VALUE。让OS决定。

max.message.bytes/message.max.bytes：消息的最大字节数，默认值1000012

rerention.ms/log.retention.ms：使用delete的日志清理策略时消息能够保留多久，默认7天。

segment.bytes/log.segment.bytes：日志分段的最大值，默认1GB。

segment.index.bytes/log.index.size.max.bytes：日志分段索引的最大值，默认10MB。

分区使用多副本来提升可靠性，但只有leader副本对外提供读写服务，而follower副本只负责在内部进行消息的同步。副本不做任何对外提供服务的操作。

本质在于数据复制，先增加新的副本，然后进行数据同步，最后删除旧的副本。

如果集群中某个主题或某个分区的流量在某段时间内特别大，那么只靠减少粒度是不足以应对的，这时需要一个限流的机制，可以对副本间的复制流量加以限制来保证重分配期间整体服务不会受太大影响。

复制限流主要有两种实现方式：kafka-config.sh脚本和kafka-reassign-partitions.sh脚本。

创建主题后可以修改分区的个数，同样可以修改副本因子（副本数）。

分区数会占用文件描述符，而一个进程所能支配的文件描述符是有限的，也就是通常所说的文件句柄的开销。

分区数过多会让Kakfa的启动和关闭耗时变长。

日志分段文件的大小超过了broker参数log.segment.bytes配置的值，默认1GB。

当前日志分段中消息的最大时间戳与系统时间戳差值大于log.roll.ms或log.roll.hours的值。默认log.roll.hours为168，7天。

偏移量文件或时间戳文件的大小达到broker端参数log.index.size.max.bytes配置的值。默认10MB。

追加的消息偏移量与日志分段的偏移量之间的差值大于Integer.MAX_VALUE。

relativeOffset：相对偏移量，消息相对于BaseOffset的偏移量，索引文件名为baseOffset的值。relativeOffset = offset - baseOffset

position：物理地址，消息在日志分段文件中对一个的物理位置。

timestamp+relativeOffset：日志分段的最大时间戳+时间戳对应的消息的相对偏移量。

根据时间戳找偏移量 -> 根据偏移量赵物理文件位置 -> 根据物理文件位置处扫描日志分段文件找到消息

基于时间的保留策略：默认保留7天（根据日志分段的最大时间戳计算）。

基于日志大小的保留策略：检测日志大小是否超过设定的阈值来删除日志分段的文件，阈值log.retention.bytes默认-1表示无穷大（所有日志文件的总大小）。log.segment.bytes单个日志分段的大小，默认1GB。

基于日志起始偏移量的保留策略：某日志分段的下一个分段的起始偏移量baseOffset是否小于等于logStartOffset，如果是，则删除此日志分段。

子主题

follower副本从leader副本拉取到消息，更新自己的LEO。

与此同时更新HW，更新HW：MIN（当前LEO，leader副本传送过来的HW值）。

Kafka

优点：让从节点分担主节点的负载压力。对于数据写压力很大而读压力很小的情况，从节点只能分摊很少的负载压力。

主题、分区号、消息头部、值、消息的时间戳

以主题为单位进行归类，而Key可以让消息再进行二次归类，同一个Key的消息会被划分到同一个分区中。

有Key的消息支持日志压缩功能。value是消息体，一般不为空，如果为空表示特定消息——墓碑消息。

KafkaProducer的3个配置参数：bootstrap.servers（生产者客户端连接Kafka集群的broker地址清单）、key.serializer和value.serializer（broker端接收的消息必须以字节数据的形式存在）

KafkaProducer是线程安全的，可以在多个线程中共享单个KafkaProducer实例，也可以将KafkaProducer实例进行池化来供其他线程调用。

发送消息的模式：发后即忘（fire-and-forget）、同步（sync）、异步（async，Callback()回调函数）。

同一个分区，如果msg1比msg2之前先发送，那么KafkaProducer就可以保证对应的callback1在callback2之前调用，回调函数的调用也可以保证分区有序。

序列化：生产者需要用序列化器把对象转换成字节数组才能通过网络发送给Kafka，消费者要使用反序列化器。

分区器：消息在发往broker的过程，有可能需要经过拦截器（Interceptor）、序列化器（Serializer）和分区器（Partitioner）的一系列作用才能被真正地发往broker。拦截器一般不是必须的。分区器的作用就是为消息分配分区。

如果key不为null，那么默认的分区器会对key进行哈希（采用MurmurHash2算法，具备高运算性能及低碰撞率），拥有相同的key会被写入同一个分区。
如果key为null，那么消息将会以轮询的方式发往主题内的各个可用分区。

如果key为null，计算得到的分区号是所有分区中的任意一个；如果key不为null，计算得到的分区号为可用分区中的任意一个。

生产者拦截器：消息发送前做一些准备工作，比如按照某个规则过滤不符合要求的消息、修改消息的内容等，也可以用来在发送回调逻辑前做一些定制化的需求，比如统计类工作。

主线程：KafkaProducer -> 拦截器 -> 序列化器 -> 分区器 -> 消息累加器（每个分区有一个消息双端队列）

Sender线程：创建Request -> Selector -> 等待Response -> 清理消息累加器

整个生产者客户端由两个线程协调运行，分别为主线程和Sender线程。主线程由KafkaProducer创建消息，然后通过可能的拦截器、序列化器和分区器的作用后缓存到消息累加器中。Sender线程负责从RecordAccumulator中获取消息并将其发送到Kafka中。

消息累加器主要用来缓存消息以便Sender线程可以批量发送，进而减少网络传输的资源消耗以提升性能。RecordAccumulator缓存的大小可以通过生产者客户端参数 buffer.memory 配置，默认32MB。如果生产者发送消息的速度超过发送到服务器的速度，则会导致生产者空间不足（这时KafkaProducer的send方法要么被阻塞，要么抛异常）。

主线程发送过来的消息都会被追加到RecordAccumulator的某个双端队列中，在消息累加器的内部为每个分区都维护了一个双端队列。

元数据是指Kafka集群的元数据，记录了集群中有哪些主题，有哪些分区，每个分区leader在哪个节点上。

InFlightRequests（发送请求）可以获得leastLoadedNode，即所有Node中负载最小的那一个。未确认的请求越多则认为负载越大。

leastLoadedNode是用于元数据请求、消费者组播协议的交互场景。

ack=1：只要分区的leader副本成功写入消息，它就会收到来自服务端的成功响应。

ack=0：生产者发送完消息不需要等待任何服务端的响应。

ack=-1或ack=all：发送消息后，需要等待ISR中的所有副本都成功写入消息之后才能收到来自服务端的成功响应。

配置消费者客户端参数及创建响应的消费者实例。

订阅主题。

拉取消息并消费。

提交消费位移。

关闭消费者实例。

Kafka消费者客户端KafkaConsumer中的4个参数，如下：

bootstrap.servers：指定连接Kafka集群所需的broker地址清单。

group.id：消费者隶属的消费组的名称（一般设置具有业务意义的名称）。

key.deserializer和value.deserializer：与生产者的中的key和value的序列化对应。

一个消费者可以订阅一个或多个主题。

如果消费者采用正则表达式的方式订阅，在之后如果有人创建了新的主题，并且名字和正则表达式相匹配，那么消费者可以消费到新添加的主题中的消息。

消费者还可以直接订阅某些主题的特定分区（KafkaConsumer的assign方法）。

Kafka消费者的订阅状态：集合订阅的方式（subscribe(Collection)）、正则表达式订阅的方式（subscribe(Pattern)）、指定分区订阅（assign(Collection)）的的方式。这三种状态是互斥的。

通过subscribe方法订阅分区是具备消费者自动均衡的，而assign不具备。

反序列化器：ByteBufferDeserializer、ByteArrayDeserializer、BytesDeserializer、DoubleDeserializer、FloatDeserializer、IntegerDeserializer、LongDeserializer、ShortDeserializer、StringDeserializer。

实际应用中，在Kafka提供的序列化器和反序列化器满足不了应用需求的时候，推荐使用Avro、JSON、Thift、ProtoBuf或Protostuff等序列化工具来包装。

Kafka中的消息是基于拉模式的。

消息的消费模式：推模式、拉模式。

Kafka消息消费是一个不断轮询的过程，消费者所要做的就是重复地调用poll方法，而poll方法返回的是所订阅的主题上的一组消息。

ConsumerRecord类属性：主题、分区、偏移量、时间戳、TimestampType（两种类型，消息创建时间、消息存储时间）、Key和value经过序列化后的大小、消息的头部内容headers、CRC32的校验值。

偏移量：对于分区而言，每条消息都有唯一的offset，用来表示消息在分区中对应的位置。

消费位移：对于消费者而言，也有一个offset，消费者使用offset表示消费到分区中某个消息所在的消息。

对于一条消息而言，偏移量和消费者消费它的消费位移是相等的。

在旧的消费者客户端中，消费位移是存储在ZooKeeper中的。而在新消费者客户端中，消费位移存储在Kafka内部的主题__consumer_offsets中。

把消费位移存储起来的动作称为 "提交"，消费者在消费完消息之后需要执行消费位移的提交。

当前消费者消费到的位置为x，那么消费者需要提交的消费位移并不是x，而是x+1。

消息丢失：当前一次拉取消息集[x+2,x+7]，如果拉取消息后就进行位移提交，即x+8，那么当前消费x+5遇到异常，故障恢复后，重新拉区消息是从x+8开始的。

重复消费：位移提交的动作是在消费完所有拉取到的消息之后才执行的。当消费x+5遇到异常，在故障恢复后，又重新拉区的消息是从x+2开始的。x+2到x+4之间的消息又重新消费了一遍。

自动位移提交的方式在正常情况下不会发生消息丢失或重复消费的现象。

Kafka提供手动位移提交的方式，手动的提交方式可以让开发人员根据程序逻辑在合适的地方进行位移提交。

enable.auto.commit：false。

手动提交分为同步提交和异步提交。

同步提交：对拉取到的每一条消息做相应的逻辑处理，或批量处理+批量提交的方式。也有可能出现重复消费问题（消费位移提交前程序突然崩溃）。commitSync()方法会根据poll方法拉取的最新位移来进行提交，只要没有发生不可恢复的错误，它就会阻塞消费者线程直至位移提交完成。

异步提交：commitAsync()方法在执行的时候消费者线程不会被阻塞，可能在提交消费位移的结果还未返回之前就开始了新一次的拉取操作。异步提交可以使消费者的性能得到一定的增强。

重复消费：commitAsync提交的时候也会失败。一般能想到的方法是重试。如果某一次提交的消费位移为x，但是提交失败了，然后下一次又异步提交了消费位移为x+y，这次成功了。如果这时重试第一次提交的消费位移x，那么此时的消费位移又变为了x。这样就会引入重复消费问题。

解决方法：设置一个递增的序号来维护异步提交的顺序，每次位移提交之后就增加序号对应的值。遇到位移提交失败需要重试的时候，可以检查提交的位移和序号的值的大小，如果前者小于后者，说明有更大的位移已经提交了；如果相等，进行重试提交。

KafkaConsumer提供了对消费速度进行控制的方法，有时我们需要暂停对某些分区的消费而先消费其他分区，当达到一定条件时再恢复这些分区的消费。

KafkaConsumer是非线程安全的，调用wakeup方法后可以退出poll的逻辑。跳出循环后一定要显示地执行关闭动作以释放运行过程中占用的各种系统资源。

当一个新的消费组建立的时候，或者消费组内的一个新消费者订阅了一个新的主题，或者__consumer_offsets主题中关于这个消费组的位移信息过期而被删除后，没有可以查找的消费位移。

Kafka中消费者找不到记录的消费位移时，会根据消费者客户端参数auto.offset.reset的配置决定从何处开始进行消费，默认值"latest"，表示从分区末尾开始消费消息。参数"earliest"表示消费者从0开始消费。参数"none"抛异常（前提是找不到消费位移）。

此时用一个新的消费组来消费某个主题时，客户端会报出重置位移的提示信息。Resetting  offset  for  partition。

seek方法指定某个分区的位移消费

指分区的所属权从一个消费者转移到另一个消费者的行为，它为消费组具备高可用性和伸缩性提供保障，使我们既方便又安全地删除消费组内的消费者或往消费组内添加消费者。

再均衡期间，消费组内的消费者是无法读取消息的。

为了防止重复消费问题，一般情况下，应该避免不必要的再均衡发生。当然也可以通过异步提交消费位移的操作避免重复消费。

主要在消费到消息或在提交消息位移时进行一些定制化的操作。

KafkaConsumer会在poll方法返回之前调用拦截器的onConsume方法来对消息进行定制化操作，比如修改返回的消息内容、按照某种规则过滤消息。

KafkaConsumer会在提交完消费位移之后调用拦截器的onCommit方法，可以使用这个方法来记录跟踪所提交的位移信息。

某些业务场景中会对消息设置一个有效期的属性，如果某条消息在既定的时间窗口内无法到达，那么就会被视为无效。自定义拦截器ConsumerInterceptorTTL实现ConsumerInterceptor<>接口。

KafkaConsumer定义了一个acquire方法，用来检测当前是否只有一个线程在操作，若有其他线程在操作则会抛出ConcurrentModifcationException异常。

acquire通过线程操作计数标记的方式检测线程是否发生了并发操作，以此保证只有一个线程在操作。acquire()和release()表示加锁和解锁。

多个消费线程同时消费同一个分区，通过assign、seek 方法实现。这种实现对于位移提交和顺序控制的处理会变得非常复杂。实际应用使用的很少。

将处理消息模块改成多线程的实现方式。

fetch.min.bytes：配置Consumer在一次拉取请求中能从Kafka中拉取的最小数据量，默认1B。

fetch.max.bytes：配置Consumer在一次拉取请求中从Kafka拉取的最大数据量。默认值50MB。如果此值比Kafka任何一条消息都小，也是可以消费的。

fetch.max.wait.ms：指定Kafka的等待时间。

max.partition.fetch.bytes：配置从每个分区返回给Consumer的最大数据量，默认值1MB。

max.poll.records：Consumer在一次拉取请求中拉取的最大消息数，默认值500条。

connection.max.idle.ms：在多久之后关闭闲置的连接，默认值9分钟。

exclude.internal.topics：Kafka的两个内部主题，__consumer_offsets和__transaction_state。exclude~指定Kafka中的内部主题是否可以向消费者公开，默认true。

receive.buffer.bytes：设置Socket接收消息缓冲区（SO_RECBUF）的大小，默认64KB。-1表示使用系统的默认值。

send.buffer.bytes：设置发送消息缓冲区（SO_SNDBUF）的大小，默认值128KB。

request.timeout.ms：Consumer等待请求响应的最长时间，默认30s。

metadata.max.age.ms：配置元数据的过期时间，默认值5分钟。

reconnect.backoff.ms：配置尝试重新连接指定主机之前的等待时间，避免频繁地连接主机，默认50ms。

retry.backoff.ms：配置尝试重新发送失败的请求到指定的主题分区之前的等待时间，避免在某些故障情况下频繁地重复发送，默认100ms。

isolation.level：配置消费者的事务隔离级别。默认 "read_uncommitted"，可以消费到HW处的位置。"read_committed"，可以消费到LSO处的位置。