7.分区中的所有副本(leader+follower)统称为

AR

(Assigned Replicas),所有与leader副本保持一定程度同步的副本（包括leader）组成

ISR

（In-Sync Replicas）.与leader副本同步滞后过多的副本（不包过leader）组成

OSR

(out-of-Syn Replicas).

leader副本负责维护和跟踪ISR集合中所有的follower副本的滞后状态，当follower滞后太多或者失效时，leader将其从ISR中剔除。如果OSR中有follower副本追上，那么从OSR转移到ISR. 默认情况下，当leader发生故障，只有ISR中的副本才有资格被宣威leader。

ISR与HW和LEO也有密切的关系。HW-Hight Watermark的缩写。高水位。他表示了一个特定的消息偏移量offest,消费者只能拉取到这个offest之前的消息。

LEO

为Log End Offest缩写。（分区中当前日志文件一条待写入消息的offest）

分区中消息是从Log Start Offest（为0）开始，到LogEndOffest结束。

HW

就是所有ISR集合中LogEndOffest的最小值

2 生产者

2.1客户端开发

一个正常的生产逻辑需要具备以下几个步骤：

配置生产者客户端参数以及创建相应的生产者实例。
构建待发送的消息
发送消息
关闭生产者实例

//生产者实例 是线程安全的
KafkaProducer<String,String> prodcuer = new KafkaProducer<>(propos);


发送的消息类。
public class ProducerRecord<K, V> {
    //主题
    private final String topic;
    //分区号
    private final Integer partition;
    //消息头
    private final Headers headers;
    //消息key
    private final K key;
    //消息值
    private final V value;
    //消息的时间戳
    private final Long timestamp;

2.1.1 必要参数

bootstrap.servers 用来指定连接Kafka集群所需的broker地址清单，多个用逗号分割。
key.serializer
value.serializer :borker端接收的消息必须以字节数组形式存在。

2.1.2 消息的发送

发送消息有三种方式：

fire-and-forget 发后即忘只管发送，不管是否到达
sync 同步
async异步

try{
  Future<RecordMetedata> future =producer.send(record);
  RecordMetedata metedata = future.get();
  //可以通过get方法来阻塞等待Kafka的响应，直到消息发送成功
}catch(ExecutionException | InterruptedException e){
  e.printStackTrace();
}

2.1.3 序列化

生产者使用序列化把对象转为字节数组才能发送给kafka, 消费者需要用对应的反序列化将字节数组转化为对象。

2.1.4 分区器

消息在通过send方法发往broker过程中，有可能需要经过拦截器（Interceptor）、序列化器（Serializer）和

分区器

（Partitioner）的一系列作用之后才能被真正的发往broker。

作用：

为消息分配分区。（没有指定分区的时候分区器会指定一个）

默认分区器是org.apache.kafka.clients.producer.internals.DefaultPartitioner。其中partition用来计算分区号，返回值为int类型。Partitioner是DefaultPartitioner的父类接口，继承了Configurable接口，通过该接口中的configure方法获取配置信息。

    /**
     * Compute the partition for the given record.
     *
     * @param topic         The topic name
     * @param numPartitions The number of partitions of the given {@code topic}
     * @param key           The key to partition on (or null if no key)
     * @param keyBytes      serialized key to partition on (or null if no key)
     * @param value         The value to partition on or null
     * @param valueBytes    serialized value to partition on or null
     * @param cluster       The current cluster metadata
     */
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster,
                         int numPartitions) {
        if (keyBytes == null) {
            return stickyPartitionCache.partition(topic, cluster);
        }
        // hash the keyBytes to choose a partition
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }

如果key不为null,那么对key进行hash，然后计算分区号。相同的key写入同一个分区。
如果key为null,那么消息会随机的方式发往主题内的任意一个分区。

2.1.5 生产者拦截器

可以在发送消息前做一些准备工作，比如过滤不合要求的消息，修改消息内容。也可以做一些定制化的需求，比如统计类工作。

使用：自定义实现ProducerInterceptor接口。需要实现3个接口

public interface ProducerInterceptor<K, V> extends Configurable {
   public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);
   public void onAcknowledgement(RecordMetadata metadata, Exception exception);
   public void close();
}

KafkaProducer在将消息序列化和计算分区前会调用生产者拦截器的onSend方法来对消息进行相应的定制化操作。

KafkaProducer会在消息被应答（Acknowledge）之前或者消息发送失败时调用onAcknowledgement方法，优先与用户设定的CallBack之前执行。

2.2 整体架构

整个生产者客户端由两个线程协调运行，这2个线程分别为主线程和Sender线程。

主线程中由KafkaProducer创建消息，通过可能的拦截器、序列化器、分区器的作用后缓存消息到消息累加器：RecordAccumulator。Sender线程从RecordAccumulator中获取消息并将期发送到Kafka中。

sender线程也是在构造函数里启动的

this.sender = newSender(logContext, kafkaClient, this.metadata);
String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
this.ioThread.start();

2.2.1 RecordAccumulator

//在KafkaProducer的构造方法中初始化            
this.accumulator = new RecordAccumulator(logContext,
                    config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),  //batchSiz  默认16384
                    this.compressionType, //消息压缩方式：none（不压缩）、gzip、snappy、lz4、zstd、
                    lingerMs(config), //用来
                    retryBackoffMs,
                    deliveryTimeoutMs,
                    metrics,
                    PRODUCER_METRIC_GROUP_NAME,
                    time,
                    apiVersions,
                    transactionManager,
                    new BufferPool(this.totalMemorySize, config.getInt(ProducerConfig.BATCH_SIZE_CONFIG), metrics, time, PRODUCER_METRIC_GROUP_NAME));

batchSize :初始化MemoryRecords实例时分配的大小
CompressionType:压缩消息的方式，默认none，不压缩
lingerMs, 用来指定生产者发送ProducerBatch之前需要等待更多消息ProducerRecord加入ProducerBatch的时间。默认为0。
retryBackoffMs：配置生产者重试的次数，默认为0。异常情况下会重试。
deliveryTimeoutMs：

requestTime(Producer请求等待响应的最长时间) + lingerMs <= deliveryTimeoutMs(传递超时时间)
metrics:
transactionManager:事务管理
BufferPool：字节缓存池

属性：

ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches;

以TopicPartition为key, 双端队列为value。收集器可以收集不同分区的消息，各自分区下有一个队列，队列中有多个ProducerBatch，每个ProducerBatch中可以放很多消息。

发送消息追加后，还会在IncompleteBatches中缓存起来，直到被ACK确认。

2.2.2 Sender线程

sender线程中维护了确认机制：

acks = 1 发送消息，leader副本写入消息，就会响应成功。默认
0：生产者发送消息，不需要响应。
-1或null，发送消息，需要ISR都写入成功才响应成功。

发送方式：每次线程启动只发送一次。 Producer new完之后，发送消息完成需要调用close方法，会关闭sender线程。

3 消费者

3.1 消费者和消费者组

消费者（Consumer）：负责订阅Kafka中的主题（Topic）,并从订阅的主题上拉取消息。
消费者组（Consumer Group）：每个消费者都有一个消费者组，消息发布到主题后，只会被投递给订阅他的每个消费者组中的其中

一个

消费者。

注意：同一个topic下会有很多的分区，同一个消费组中可以增加消费者来让消费能力提升。但是当消费者过多，就会导致有的消费者分配不到任何分区。

分配策略是通过消费这客户端参数partition.assignment.strategy来配置的。

3.1.1 消息投递模式

点对点（p2p）如果所有消费者都属于同一个消费组。那么所有消息就会被均衡的投递到每一个消费者。每条消息只会被一个消费者消费。
发布订阅：如果所有消费者都属于不同的消费组。那么所有的消息都会被广播给所有的消费者。那么一个消息会被所有消费者处理。

3.2 客户端开发

一个正常的客户端需要一下几个步骤：

配置消费者客户端参数及创建相应的消费者实例。
订阅主题
拉取消息
提交消费位移
关闭消费者实例。

3.2.1 必要参数

bootstrap.servers。和生产者中一样
group.id: 消费者组的名称，默认为“”。如果为空，会报错。一般会设置成具有一定业务意义的名称。
key.deserializer 和 value.deserializer:用于反序列化。

3.2.2 订阅主题与分区

订阅方式

：AUTO_TOPICS:集合订阅的方式 AUTO_PATTERN:正则订阅方式 USER_ASSIGNED:assign方式

三种方式互斥，一个消费者只能使用一种。否则会报错。

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

consumer.subscribe(Arrays.asList(topic1));
consumer.subscribe(Arrays.asList(topic2));
consumer.subscribe(Pattern.complie("topic-.*"));

可以使用集合或者正则表达式的形式订阅特定模式的主题。如果前后2次订阅了不通的主题，以最后一次为准。

2.消费者还可以通过KafkaConsumer中的assign()方法订阅主题主题中特定的分区。

public void assign(Collection<TopicPartition> partitions)


//通过该方法可以获取主题下的所有分区信息 （包括 AR ISR OSR集合）
public List<PartitionInfo> partitionsFor(String topic, Duration timeout)

有以上方法，所以我们可以通过assign方法也能实现订阅主题（全部分区）的功能。

3.消费者可以通过unsubscribe()方法来取消主题的订阅。

3.2.3 反序列化

生产者使用序列化把对象转为字节数组才能发送给kafka, 消费者需要用对应的反序列化将字节数组转化为对象。

都可以自定义序列化方式，不过生产者和消费者得配对。

3.2.4 消费消息

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

//拉取消息方法
public ConsumerRecords<K, V> poll(final Duration timeout)

Kafka消费消息是一个不断循环拉取的过程，也就是重复的调用poll方法。poll方法是所订阅主题（分区）上的一组消息。

3.2.5 位移提交

消费者中的offest来表示消费到分区中某个消息所在的位置。需要持久化保存，不然重启后，无法知道消费到哪个位置。

//获取消费位置(position)
public long position(TopicPartition partition)
//获取已经提交过的消费位移（committed Offset）
public OffsetAndMetadata committed(TopicPartition partition)

position = committed offest = lastConsumedOffset +1

位移提交时机：

消费者消费获取一批消息，如果消费一部分，然后异常导致没有位移提交，就会导致重复消费。

位移提交方式

自动提交（默认）：enable.auto.commit配置为true，然后定期提交。auto.commit.interval.ms配置周期，默认5s.

缺点：重复消费、消息丢失问题优点：编码简单

手动提交：enable.auto.commit配置为false。

同步提交：commitsync 可以按分区提交

异步提交：commitAsync 可以增加回调方法

 public void commitSync()
 public void commitSync(Duration timeout)
 public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets)
 public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets, final Duration timeout)
   
 public void commitAsync()
 public void commitAsync(OffsetCommitCallback callback)
 public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)

异步提交回调函数失败，如果重试，会有先后问题，导致重复消费。

3.2.6 控制或关闭消费

Kafka提供了对消费速度进行控制的方法。通过pause()和resume()方法来分别实现暂停和恢复。

3.2.7 指定位移消费

新消费者加入时，没有可以查找的消费位移。配置auto.offset.restart可以在找不到消费位移时决定从何处开始消费

latest 从分区末尾开始，也就是下一条
earliest:从0开始
none:找不到时抛出异常。

以上只是找不到时的处理。seek可以指定位移：

public void seek(TopicPartition partition, long offset)

seek只能重置分区的消费位置，而拉取哪个分区的消息是poll中实现的，所以seek之前必须要先poll

通过seek可以跳过或者回溯消息。

3.2.8 再均衡

分区的所有权从一个消费者转移到另一个消费者的行为。

优点：高可用，伸缩性。可以安全的删除消费组内的消费者或者添加新的消费者。

缺点：

再均衡过程中，消费组不可用。
消费者状态丢失。比如：消费者还没提交消费位移的时候，发生再均衡，会导致重复消费。

public void subscribe(Collection<String> topics)
public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener)
public void subscribe(Pattern pattern)
public void subscribe(Pattern pattern, ConsumerRebalanceListener listener)

ConsumerRebalanceListener:再均衡监听器，用来设定再均衡动作前后的一些准备和收尾动作。

public interface ConsumerRebalanceListener {
  //会在再均衡开始之前和消费者停止读取消息之后被调用 partitions：重分配前
  void onPartitionsRevoked(Collection<TopicPartition> partitions);
  //在重新分配分区之后和消费者开始读取消息之前被调用。partitions：重分配后
  void onPartitionsAssigned(Collection<TopicPartition> partitions);

3.2.9 消费者拦截器

Kafka会在poll方法返回结果之前，调用拦截器的onConsume方法，对消息进行定制化的操作。

public interface ConsumerInterceptor<K, V> extends Configurable, AutoCloseable {
  //在poll方法返回结果前
  public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records);
  //在提交完消费位移之后。
  public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets);
  public void close();

3.2.10 多线程实现

生产者是现成安全的，但是消费者不是。

//KafkaConsumer中
//通过这个方法判断是不是只有一个线程在操作。    相当与一个锁，将refcount计数+1
private void acquire() {
        long threadId = Thread.currentThread().getId();
        if (threadId != currentThread.get() && !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId))
            throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access");
        refcount.incrementAndGet();
    }
//释放锁
    private void release() {
        if (refcount.decrementAndGet() == 0)
            currentThread.set(NO_CURRENT_THREAD);
    }

实现方式：使用滑动窗口

一个方格代表一个批次的消息，一个滑动窗口包含若干方法，startOffset滑动窗口开始位置，endOffset结束位置，

每当startOffset中的消息被消费完成，就能提交这部分位移，窗口向前滑动一步。

一个方格代表一个线程，如果startOffset无法被消费完成，悬停一定时间后就可以重试，重试失败就转入重试队列，再不行就进入死信队列。

3.2.11 重要消费参数

fetch.min.bytes: 拉取请求中能从Kafka中拉取的最小数据量。如果小于该值，会进行等待。
fetch.min.bytes: 拉取的最大数据量。
fetch.max.wait.ms：与fetch.min.bytes参数相关，防止一直等待。

4 主题与分区

4.1 优先副本的选举

优先副本：AR集合的第一个副本[1,2,0] 优先副本为1。

4.2 文件目录

一个主题有很多分区，分区有很多副本，一个副本对应一个目录，目录下主要有

三类文件。 *.index *.log *.timeindex

4.3 日志索引

Kafka索引文件以稀疏索引的方式构造消息的索引。每当写入一定量的消息时，偏移量索引文件和时间戳索引文件分别增加一个偏移量索引项和时间戳索引项。

在索引中使用二分查找法。

4.4 日志快速读取

日志删除：

日志压缩：相同key的value，只保留最新

磁盘存储：文件只允许追加，不允许修改。其实是顺序写磁盘的一种，加快了速度。

页缓存：Kafka不使用Java虚拟机缓存数据，使用页缓存。Jvm gc会变慢。

零拷贝：应用程序直接请求内核磁盘中的数据传输给socket.

5、队列

就是日志存储，按顺序存储，然后按位移消费。

原文链接：https://blog.csdn.net/LookOutThe/article/details/120998568

1 初识KafKa

1.1基本概念