当前位置：首页 > news >正文

Kafka核心概念与源码阅读

news 来源：原创 2024/4/20 16:16:39

Kafka学习笔记

Kafka简介

More than 80% of all Fortune 100 companies trust, and use Kafka.

Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications.

Apache Kafka is the most popular open-source stream-processing software for collecting, processing, storing, and analyzing data at scale. Most known for its excellent performance, low latency, fault tolerance, and high throughput, it’s capable of handling thousands of messages per second. With over 1,000 Kafka use cases and counting, some common benefits are building data pipelines, leveraging real-time data streams, enabling operational metrics, and data integration across countless sources.

这段简介摘自Kafka的官网描述。

Kafka对自己的定位是： open-source distributed event streaming platform，开源的分布式消息引擎系统。从简介可以看出，它不仅仅是消息队列，它在大数据领域的场景更友好。

Kafka 使用 Scala 和 Java 语言开发，设计上大量使用了批量和异步的思想，这种设计使得 Kafka 能做到超高的性能。Kafka 的性能，尤其是异步收发的性能，是三者中最好的，但与 RocketMQ 并没有量级上的差异，大约每秒钟可以处理几十万条消息。

Kafka的特点

第三段描述中，Kafka是这样描述自己的：最流行的开源流处理平台，适用于大规模收集、处理、存储与分析数据。Kafka有着卓越的性能、低延迟、容错量和高吞吐量。

Kafka模型

Kafka是采用了发布-订阅模型，先看下官网的模型示意图：

在这里插入图片描述

在图上，topic中有p1-p4共四个分区，两个不同的生产者客户端，它们将数据传送到不同的分区中，图中相同颜色的键代表同一类型的事件，同一类型的事件被写入了同一个分区。如果场景合适，两个生产者也可以都把事件发送到同一分区中。

Kafka的几个概念

消息：Record。消息就是指 Kafka 处理的主要对象。
主题：Topic。主题是承载消息的逻辑容器，在实际使用中多用来区分具体的业务。
分区：Partition。一个有序不变的消息序列。每个主题下可以有多个分区。
消息位移：Offset。表示分区中每条消息的位置信息，是一个单调递增且不变的值。
副本：Replica。Kafka 中同一条消息能够被拷贝到多个地方以提供数据冗余，这些地方就是所谓的副本。副本还分为领导者副本和追随者副本，各自有不同的角色划分。副本是在分区层级下的，即每个分区可配置多个副本实现高可用。

副本的工作机制也很简单：生产者总是向领导者副本写消息；而消费者总是从领导者副本读消息。至于追随者副本，它只做一件事：向领导者副本发送请求，请求领导者把最新生产的消息发给它，这样它能保持与领导者的同步。
生产者：Producer。向主题发布新消息的应用程序。
消费者：Consumer。从主题订阅新消息的应用程序。
消费者位移：Consumer Offset。表示消费者消费进度，每个消费者都有自己的消费者位移。
消费者组：Consumer Group。多个消费者实例共同组成的一个组，同时消费多个分区以实现高吞吐。
重平衡：Rebalance。消费者组内某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程。Rebalance 是 Kafka 消费者端实现高可用的重要手段。

Kafka生产者

Kafka的生产者要做的就是一件事：将消息发送给Kafka。

API使用

使用demo：（demo摘自KafkaProducer类注释）

	   //属性配置
	   Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		//new Producer
       Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++)
            //producer.send 发送消息给Kafka
            producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));
		//关闭Producer
        producer.close();

源码阅读

先看下KafkaProducer这个类的构造函数中我认为几个比较重要的属性设置

		//事务id
        String transactionalId;
        //日志上下文
        LogContext logContext;
		//从Properties中拿出拦截器配置
	    List<ProducerInterceptor<K, V>> interceptorList = (List) configWithClientId.getConfiguredInstances(
                    ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptor.class);
        //拦截器设置
        if (interceptors != null)
            this.interceptors = interceptors;
        else
            this.interceptors = new ProducerInterceptors<>(interceptorList);
        //计数器
        this.accumulator = new RecordAccumulator();
        //发送者
        this.sender = newSender(logContext, kafkaClient, this.metadata);
       //Kafka线程
        this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
        this.ioThread.start();

从构造函数我们知道：

KafkaProducer有事务相关的能力
KafkaProducer支持一组拦截器
KafkaProducer有个累加器，结合之前在消息队列中说过，Kafka通过批量发送来提升性能，那不难猜出累加器就是用于批量发送的一个计数器
newSender
KafkaThread是真正发送的线程，并且它的Runnable又是newSender，并且它是个守护者线程

那我们首先可以得出一个结论，在开发应用中，最好不要发送一个事件就new一个KafkaProducer，会创建N多线程，实际上一个模块中，同样的发送配置（指的是消息发往同一个topic的同一个partition这种）只有一个公共的Producer就可以啦。

其次，Kafka支持拦截器，那么我们就可以实现自己的Producer拦截器，并且支持多个，在属性配置Properties的时候指定即可

并且，阅读KafkaProducer的构造方法源码也可以看出，Kafka也确实是批量发送消息的，而不是调用一次send方法就发送一次消息，所以Kafka也确实不适合需要在线处理的场景。

接下来来看KafkaProducer的send方法，KafkaProducer有两个send方法，分别是：

 public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
     	//Kafka Producer 是异步发送消息的，这个方法会立即返回，但此时你不能认为消息发送已成功完成。
        return send(record, null);
    }
  public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
        // 设置拦截器的onsend方法，拦截器的方法调用要早于callback 的调用，并且因为send方法实际上是守护线程执行的，所以拦截器的方法和callback不是在同一个线程中执行的
        ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
        return doSend(interceptedRecord, callback);
    }

再看下doSend方法核心逻辑：

private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
        TopicPartition tp = null;
        //生产者检查
        throwIfProducerClosed();
        // 首先保证元数据中的topic有效
        KafkaProducer.ClusterAndWaitTime clusterAndWaitTime;
        try {
            clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
        } catch (KafkaException e) {
            if (metadata.isClosed())
                throw new KafkaException("Producer closed while send in progress", e);
            throw e;
        }
            ...
        //分区
        int partition = partition(record, serializedKey, serializedValue, cluster);
        tp = new TopicPartition(record.topic(), partition);
        //序列化日志等相关
            ...
        // 如果有指定的callback，调用拦截器的回调与指定callback
        Callback interceptCallback = new KafkaProducer.InterceptorCallback<>(callback, this.interceptors, tp);
        //事务设置
        if (transactionManager != null && transactionManager.isTransactional()) {
            transactionManager.failIfNotReadyForSend();
        }
        //计数器
        RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey,
                serializedValue, headers, interceptCallback, remainingWaitMs, true);
        //如果之间待发送的消息批次已经满了，就会重新分区，重复上面的操作
		...
        //事务相关
            ...
        //如果待发送的消息已经满足条件了，就唤醒守护者线程去发送消息给Kafak
        if (result.batchIsFull || result.newBatchCreated) {
            log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);
            this.sender.wakeup();
        }
        return result.future;
    }

分区策略

所谓分区策略是决定生产者将消息发送到哪个分区的算法

我们继续上面的源码，看下partition方法：

private final Partitioner partitioner; 
private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
        Integer partition = record.partition();
        return partition != null ?
                partition :
                partitioner.partition(
                        record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
    }

那我们就可以通过实现Partitioner接口来实现自己的分区策略，当然Kafka也有自己默认的分区策略：
在这里插入图片描述

DefaultPartitioner默认分区策略

关键代码：

//hash the keyBytes to choose a partition 
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;

这种方法是通过哈希计算出的，所以相同的key分区一定相同

RoundRobinPartitioner轮询策略

也称 Round-robin 策略，即顺序分配。比如一个主题下有 3 个分区，那么第一条消息被发送到分区 0，第二条被发送到分区 1，第三条被发送到分区 2，以此类推。当生产第 4 条消息时又会重新开始，即将其分配到分区 0，

轮询策略有非常优秀的负载均衡表现，它总是能保证消息最大限度地被平均分配到所有分区上，故默认情况下它是最合理的分区策略，也是我们最常用的分区策略之一。

UniformStickyPartitioner

这个类的注释中说到，这种分区方式的分区策略并没有使用到key，并且从关键代码看出它其实就是随机的选了个分区

那相同的key其实也不一定在同一个分区

		//查询可用分区  
		List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
            if (availablePartitions.size() < 1) {
                //如果没有可用分区，就随机生成个新的
                Integer random = Utils.toPositive(ThreadLocalRandom.current().nextInt());
                newPart = random % partitions.size();
            } else if (availablePartitions.size() == 1) {
                //如果恰巧只有一个分区，就直接使用
                newPart = availablePartitions.get(0).partition();
            } else {
                //如果可用分区数量>1并且当前分区依旧为null
                while (newPart == null || newPart.equals(oldPart)) {
                    //从可用分区中随便找一个使用
                    Integer random = Utils.toPositive(ThreadLocalRandom.current().nextInt());
                    newPart = availablePartitions.get(random % availablePartitions.size()).partition();
                }
            }

实现我们自己的分区策略

首先看下官方给出的demo：SamplePartition

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        /*
            key-1
            key-2
            key-3
         */
        String keyStr = key + "";
        String keyInt = keyStr.substring(4);
        System.out.println("keyStr : "+keyStr + "keyInt : "+keyInt);

        int i = Integer.parseInt(keyInt);

        return i%2;
    }

那首先我们要做的就是实现org.apache.kafka.clients.producer.Partitioner接口。这个接口也很简单，只定义了两个方法：partition()和close()，通常你只需要实现最重要的 partition 方法。我们来看看这个方法的方法签名：

int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);

这里的topic、key、keyBytes、value和valueBytes都属于消息数据，cluster则是集群信息（比如当前 Kafka 集群共有多少主题、多少 Broker 等）。Kafka 给你这么多信息，就是希望让你能够充分地利用这些信息对消息进行分区，计算出它要被发送到哪个分区中。只要你自己的实现类定义好了 partition 方法，同时设置partitioner.class参数为你自己实现类的 Full Qualified Name，那么生产者程序就会按照你的代码逻辑对消息进行分区。

那我们就可以按需定制自己想要的分区策略，比如基于地理位置的分区策略：当然这种策略一般只针对那些大规模的 Kafka 集群，特别是跨城市、跨国家甚至是跨大洲的集群。

此时我们就可以根据 Broker 所在的 IP 地址实现定制化的分区策略。比如下面这段代码：

List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
return partitions.stream().filter(p -> isSouth(p.leader().host())).map(PartitionInfo::partition).findAny().get();

总结

最好不要发送一个事件就new一个KafkaProducer，会创建N多线程，实际上一个模块中，同样的发送配置（指的是消息发往同一个topic的同一个partition这种）只有一个公共的Producer就可以
其次，Kafka支持自定义拦截器，那么我们就可以实现自己的Producer拦截器，并且支持多个，在属性配置Properties的时候指定即可
Kafka也确实是批量发送消息的，而不是调用一次send方法就发送一次消息，所以Kafka也确实不适合需要在线处理的场景。
Kafka Producer 是异步发送消息的，如果你调用的是 producer.send(msg) 这个 API，那么它通常会立即返回，但此时你不能认为消息发送已成功完成。所以建议使用 producer.send(msg, callback)，否则你会调用完 producer.send(msg) 之后，误以为Kafka丢失了消息
Kafka支持自定义分区策略，默认的是用哈希计算出的

Kafka消费者

Kafka消费者要做的事情就是订阅生产者的事件，然后拿到它进行处理

API使用

使用demo：（demo摘自KafkaProducer类注释）

try {
            // 消费订阅哪一个Topic或者几个Topic
            consumer.subscribe(Arrays.asList(TOPIC_NAME));
            while(true) {
                //获取指定分区的消息数据
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
                //根据partition循环处理
                for (TopicPartition partition : records.partitions()) {
                    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                    for (ConsumerRecord<String, String> record : partitionRecords) {
                        System.out.println(record.offset() + ": " + record.value());
                    }
                    //获取本次lastOffset
                    long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                    //提交offset  
                    consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
                }
            }
        } finally {
            consumer.close();
        }

Kafka消费者组

消费者组，即 Consumer Group。用一句话概括就是：Consumer Group 是 Kafka 提供的可扩展且具有容错性的消费者机制。既然是一个组，那么组内必然可以有多个消费者或消费者实例（Consumer Instance），它们共享一个公共的 ID，这个 ID 被称为 Group ID。组内的所有消费者协调在一起来消费订阅主题（Subscribed Topics）的所有分区（Partition）。当然，每个分区只能由同一个消费者组内的一个 Consumer 实例来消费。

消费者组的三个特性：

Consumer Group 下可以有一个或多个 Consumer 实例。这里的实例可以是一个单独的进程，也可以是同一进程下的线程。在实际场景中，使用进程更为常见一些。
Group ID 是一个字符串，在一个 Kafka 集群中，它标识唯一的一个 Consumer Group。
Consumer Group 下所有实例订阅的主题的单个分区，只能分配给组内的某个 Consumer 实例消费。这个分区当然也可以被其他的 Group 消费。

当 Consumer Group 订阅了多个主题后，组内的每个实例不要求一定要订阅主题的所有分区，它只会消费部分分区中的消息。

Consumer Group 之间彼此独立，互不影响，它们能够订阅相同的一组主题而互不干涉。可以这么说，Kafka 仅仅使用 Consumer Group 这一种机制，却同时实现了传统消息引擎系统的两大模型：如果所有实例都属于同一个 Group，那么它实现的就是消息队列模型；如果所有实例分别属于不同的 Group，那么它实现的就是发布 / 订阅模型。

理想情况下，Consumer 实例的数量应该等于该 Group 订阅主题的分区总数。

位移（Offset）

消费者在消费的过程中需要记录自己消费了多少数据，即消费位置信息。在 Kafka 中，这个位置信息有个专门的术语：位移（Offset），也就是上面demo中手动提交的就是下次要消费的位移位置。

老版本的 Consumer Group 把位移保存在 ZooKeeper 中。Apache ZooKeeper 是一个分布式的协调服务框架，Kafka 重度依赖它实现各种各样的协调管理。将位移保存在 ZooKeeper 外部系统的做法，最显而易见的好处就是减少了 Kafka Broker 端的状态保存开销。现在比较流行的提法是将服务器节点做成无状态的，这样可以自由地扩缩容，实现超强的伸缩性。Kafka 最开始也是基于这样的考虑，才将 Consumer Group 位移保存在独立于 Kafka 集群之外的框架中

在新版本的 Consumer Group 中，Kafka 社区重新设计了 Consumer Group 的位移管理方式，采用了将位移保存在 Kafka 内部主题—— __consumer_offsets。

提交位移

目前 Kafka Consumer 提交位移的方式有两种：**自动提交位移和手动提交位移。**例如上面的demo中演示的，我们是通过手动提交的方式来提交位移。

Consumer 端有个参数叫 enable.auto.commit，如果值是 true，则 Consumer 在后台默默地为你定期提交位移，提交间隔由一个专属的参数 auto.commit.interval.ms 来控制。自动提交位移有一个显著的优点，就是省事，你不用操心位移提交的事情，就能保证消息消费不会丢失。但这一点同时也是缺点。因为它太省事了，以至于丧失了很大的灵活性和可控性，你完全没法把控 Consumer 端的位移管理。

另一种位移提交方式：手动提交位移，即设置 enable.auto.commit = false。一旦设置了 false，作为 Consumer 应用开发的你就要承担起位移提交的责任。Kafka Consumer API 为你提供了位移提交的方法，如 consumer.commitSync 等。当调用这些方法时，Kafka 会向位移主题写入相应的消息。

自动提交位移，有可能存在一个问题：只要 Consumer 一直启动着，它就会无限期地向位移主题写入消息。

Kafka 使用Compact 策略来删除位移主题中的过期消息，避免该主题无限期膨胀。那么应该如何定义 Compact 策略中的过期呢？对于同一个 Key 的两条消息 M1 和 M2，如果 M1 的发送时间早于 M2，那么 M1 就是过期消息。Compact 的过程就是扫描日志的所有消息，剔除那些过期的消息，然后把剩下的消息整理在一起。

Kafka 提供了专门的后台线程定期地巡检待 Compact 的主题，看看是否存在满足条件的可删除数据。这个后台线程叫 Log Cleaner。很多实际生产环境中都出现过位移主题无限膨胀占用过多磁盘空间的问题，可以检查一下 Log Cleaner 线程的状态，通常都是这个线程挂掉了导致的。

位移主题

__consumer_offsets 在 Kafka 源码中有个更为正式的名字，叫位移主题，即 Offsets Topic。

新版本 Consumer 的位移管理机制其实也很简单，就是**将 Consumer 的位移数据作为一条条普通的 Kafka 消息，提交到 __consumer_offsets 中。可以这么说，__consumer_offsets 的主要作用是保存 Kafka 消费者的位移信息。**它要求这个提交过程不仅要实现高持久性，还要支持高频的写操作。显然，Kafka 的主题设计天然就满足这两个条件，因此，使用 Kafka 主题来保存位移这件事情，实际上就是一个水到渠成的想法了。

虽说位移主题是一个普通的 Kafka 主题，但它的消息格式却是 Kafka 自己定义的，用户不能修改，也就是说你不能随意地向这个主题写消息，因为一旦你写入的消息不满足 Kafka 规定的格式，那么 Kafka 内部无法成功解析，就会造成 Broker 的崩溃。事实上，Kafka Consumer 有 API 帮你提交位移，也就是向位移主题写消息。

位移主题的 Key 中保存了 3 部分内容：<Group ID，主题名，分区号 >。

位移主题的消息格式可不是只有这一种。事实上，它有 3 种消息格式。除了刚刚我们说的这种格式，还有 2 种格式：

用于保存 Consumer Group 信息的消息。
用于删除 Group 过期位移甚至是删除 Group 的消息。

通常来说，当 Kafka 集群中的第一个 Consumer 程序启动时，Kafka 会自动创建位移主题。

Rebalance

Rebalance 本质上是一种协议，规定了一个 Consumer Group 下的所有 Consumer 如何达成一致，来分配订阅 Topic 的每个分区。比如某个 Group 下有 20 个 Consumer 实例，它订阅了一个具有 100 个分区的 Topic。正常情况下，Kafka 平均会为每个 Consumer 分配 5 个分区。这个分配的过程就叫 Rebalance。

Rebalance 的触发条件有 3 个：

组成员数量发生变化
订阅主题数量发生变化
订阅主题的分区数发生变化

Rebalance的问题

Rebalance 影响 Consumer 端 TPS。在 Rebalance 过程中，所有 Consumer 实例都会停止消费，等待 Rebalance 完成。
Rebalance 效率不高。当前 Kafka 的设计机制决定了每次 Rebalance 时，Group 下的所有成员都要参与进来，而且通常不会考虑局部性原理，但局部性原理对提升系统性能是特别重要的。
Rebalance 很慢。

Coordinator协调者

所谓协调者，在 Kafka 中对应的术语是 Coordinator，它专门为 Consumer Group 服务，负责为 Group 执行 Rebalance 以及提供位移管理和组成员管理等。Consumer 端应用程序在提交位移时，其实是向 Coordinator 所在的 Broker 提交位移。同样地，当 Consumer 应用启动时，也是向 Coordinator 所在的 Broker 发送各种请求，然后由 Coordinator 负责执行消费者组的注册、成员管理记录等元数据管理操作。

在 Rebalance 过程中，所有 Consumer 实例共同参与，在协调者组件的帮助下，完成订阅主题分区的分配。但是，在整个过程中，所有实例都不能消费任何消息，因此它对 Consumer 的 TPS 影响很大。

所有 Broker 在启动时，都会创建和开启相应的 Coordinator 组件。也就是说，所有 Broker 都有各自的 Coordinator 组件。

如何避免Rebalance

第一类非必要 Rebalance 是因为未能及时发送心跳，导致 Consumer 被“踢出”Group 而引发的。例如可以通过修改配置参数session.timeout.ms 和 heartbeat.interval.ms
第二类非必要 Rebalance 是 Consumer 消费时间过长导致的。
第三排查一下Consumer 端的 GC 表现，比如是否出现了频繁的 Full GC 导致的长时间停顿，从而引发了 Rebalance。

Kafka核心概念与源码阅读

Kafka学习笔记

Kafka简介

Kafka的特点

Kafka模型

Kafka的几个概念

Kafka生产者

API使用

源码阅读

分区策略

DefaultPartitioner默认分区策略

RoundRobinPartitioner轮询策略

UniformStickyPartitioner

实现我们自己的分区策略

总结

Kafka消费者

API使用

Kafka消费者组

位移（Offset）

提交位移

位移主题

相关配置参数

Rebalance

Rebalance的问题

Coordinator协调者

如何避免Rebalance

相关参数配置

相关文章：