kafka 生产数据命令(30分钟带你图解Kafka生产者初始化核心流程)

认真读完这篇文章,我相信你会对 Kafka 生产初始化源码有更加深刻的理解。

这篇文章干货很多,希望你可以耐心读完。

kafka 生产数据命令(30分钟带你图解Kafka生产者初始化核心流程)(1)

01 总体概述

我们都知道在 Kafka 中,我们把产生消息的一方称为生产者即 Producer,它是 kafka 核心组件之一,也是消息的来源所在。那么这些生产者产生的消息是如何传到 Kafka 服务端的呢?初始化过程是怎么样的呢?接下来会逐一讲解说明。

02 生产者初始化核心组件及流程剖析

我们先从生产者客户端构造 KafkaProducer 开始讲起:

Properties properties = new Properties(); //构造 KafkaProducer KafkaProducer producer = new KafkaProducer(properties); //调用send异步回调发送 producer.send(record,new DemoCallBack(record.topic(), record.key(), record.value()));

上面代码主要做了2件事情:

1)初始化 KafkaProducer 实例 。

2)调用 send 接口发送数据,支持同步和异步回调方式 。

待构造完 KafkaProducer 就正式进入生产者源码的入口了,如下图所示:

kafka 生产数据命令(30分钟带你图解Kafka生产者初始化核心流程)(2)

接下来我们分析一下 KafkaProducer 的源码, 先看下该类里面的 「 重要字段 」 :

public class KafkaProducer<K, V> implements Producer<K, V> { private final Logger log; private static final String JMX_PREFIX = "kafka.producer"; public static final String NETWORK_THREAD_PREFIX = "kafka-producer-network-thread"; public static final String PRODUCER_METRIC_GROUP_NAME = "producer-metrics"; // 生产者客户端Id private final String clientId; // 消息分区器 private final Partitioner partitioner; // 消息的最大的长度,默认1M,生产环境可以提高到10M private final int maxRequestSize; // 发送消息的缓冲区的大小,默认32M private final long totalMemorySize; // 集群元数据 private final ProducerMetadata metadata; // 消息累加器 private final RecordAccumulator accumulator; // 执行发送消息的类 private final Sender sender; // 执行发送消息的线程 private final Thread ioThread; // 消息压缩类型 private final CompressionType compressionType; // key的序列化器 private final Serializer<K> keySerializer; // value的序列化器 private final Serializer<V> valueSerializer; // 生产者客户端参数配置 private final ProducerConfig producerConfig; // 等待元数据更新的最大时间,默认1分钟 private final long maxBlockTimeMs; // 生产者拦截器 private final ProducerInterceptors<K, V> interceptors; // api版本 private final ApiVersions apiVersions; // 事务管理器 private final transactionManager transactionManager; ........ }

重要且核心字段含义如下:

1) clientId: 生产者客户端的ID。

2) partitioner: 消息的分区器,即通过某些算法将消息分配到某一个分区中。

3) maxRequestSize: 消息的最大的长度,默认1M,生产环境可以提高到10M。

4) totalMemorySize: 发送消息的缓冲区的大小,默认32M。

5) metadata: 集群的元数据。

6) accumulator: 消息累加器,主要负责缓冲消息。

7) sender: 执行发送消息的类型,主要负责发送消息。

8) ioThread: 执行发送消息的线程,主要负责封装Sender类。

9) compressionType: 消息压缩的类型。

10) keySerializer: key的序列化器。

11) valueSerializer: value的序列化器。

12) producerConfig :生产者客户端的配置参数。

13) maxBlockTimeMs: 等待元数据更新和缓冲区分配的最长时间,默认1分钟。

14) interceptors: 生产者拦截器。主要负责在消息发送前后对消息进行拦截和处理。

接下来我们看下KafkaProducer的构造方法,来剖析生产者发送消息的过程中涉及到的「核心组件」。

源码位置:

kafka\clients\src\main\java\org\apache\kafka\clients\producer\KafkaProducer.java 323行

如果不会安装源码环境的话,可以参考之前的 Kafka源码之旅入门篇

public class KafkaProducer<K, V> implements Producer<K, V> { ...... KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer, ProducerMetadata metadata, KafkaClient kafkaClient, ProducerInterceptors<K, V> interceptors, Time time) { // 1.生产者配置初始化 ProducerConfig config = new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, keySerializer, valueSerializer)); try { // 2.获取客户端配置参数 Map<String, Object> userProvidedConfigs = config.originals(); this.producerConfig = config; this.time = time; // 3.用于事务传递的TransactionalId,保证会话的可靠性,如果配置表示启用幂等 事务 String transactionalId = (String) userProvidedConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG); // 4.设置生产者客户端id this.clientId = config.getString(ProducerConfig.CLIENT_ID_CONFIG); LogContext logContext; // 根据事务id是否配置来记录不同日志 if (transactionalId == null) logContext = new LogContext(String.format("[Producer clientId=%s] ", clientId)); else logContext = new LogContext(String.format("[Producer clientId=%s, transactionalId=%s] ", clientId, transactionalId)); log = logContext.logger(KafkaProducer.class); log.trace("Starting the Kafka producer"); ........省略Metrics // 5.设置对应的分区器 this.partitioner = config.getConfiguredInstance(ProducerConfig.PARTITIONER_CLASS_CONFIG, Partitioner.class); // 6.失败重试的退避时间,配置参数:retry.backoff.ms 默认100ms long retryBackoffMs = config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG); // 7.定义key、value对应的序列化器 if (keySerializer == null) { this.keySerializer = config.getConfiguredInstance(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, this.keySerializer.configure(config.originals(Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, clientId)), true); } else { config.ignore(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG); this.keySerializer = keySerializer; } if (valueSerializer == null) { this.valueSerializer = config.getConfiguredInstance(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, this.valueSerializer.configure(config.originals(Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, clientId)), false); } else { config.ignore(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG); this.valueSerializer = valueSerializer; } // load interceptors and make sure they get clientId userProvidedConfigs.put(ProducerConfig.CLIENT_ID_CONFIG, clientId); ProducerConfig configWithClientId = new ProducerConfig(userProvidedConfigs, false); // 8.定义生产者拦截器列表 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); ClusterResourceListeners clusterResourceListeners = configureClusterResourceListeners(keySerializer, valueSerializer, interceptorList, reporters); // 9.设置消息的最大的长度,默认1M,生产环境可以提高到10M this.maxRequestSize = config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG); // 10.设置发送消息的缓冲区的大小,默认32M this.totalMemorySize = config.getLong(ProducerConfig.BUFFER_MEMORY_CONFIG); // 11.设置消息压缩类型 this.compressionType = CompressionType.forName(config.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG)); // 12.设置等待元数据更新的最大时间,默认1分钟 this.maxBlockTimeMs = config.getLong(ProducerConfig.MAX_BLOCK_MS_CONFIG); // 13.设置消息投递的超时时间 int deliveryTimeoutMs = configureDeliveryTimeout(config, log); this.apiVersions = new ApiVersions(); // 事务管理器 this.transactionManager = configureTransactionState(config, logContext); ....省略,看下面各小节源码 config.logUnused(); AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics, time.milliseconds()); log.debug("Kafka producer started"); } catch (Throwable t) { // call close methods if internal objects are already constructed this is to prevent resource leak. see KAFKA-2121 close(Duration.ofMillis(0), true); // now propagate the Exception throw new KafkaException("Failed to construct kafka producer", t); } } }

下面通过一张图来描述 KafkaProducer 的 初始化源码的过程:

kafka 生产数据命令(30分钟带你图解Kafka生产者初始化核心流程)(3)

Kafka Producer 初始化核心组件如下:

1)初始化生产者配置 (ProducerConfig) 。

2)设置客户端配置文件的配置信息 (userProvidedConfigs)

3)设置事务ID (transactionaID) 。

4)设置生产者客户端ID (clientId)

5)设置对应的分区器 (partitioner) , 支持自定义,用来将消息分配给某个主题的某个分区的。

6)设置失败重试的退避时间 (retryBackoffMs) 。在客户端请求服务端时,可能因为网络或服务端异常造成请求超时。这时请求失败会重试,但是如果重试的频率过高又可能造成服务端网络拥堵。因此必须等一段时间再请求,默认100ms。

7)初始化key的序列化器 (keySerializer) 和value的序列化器 (valueSerializer) 。key和value的序列化器是用户在初始化 KafkaProducer 的时候自定义的。

8)设置生产者拦截器 (ProducerInterceptor) ,拦截器的主要作用是按照一定的规则统一对消息进行处理。

9)设置 消息的最大的长度 (maxRequestSize) 。默认是1M,超了会报异常。在生产环境中建议设置为10M。

10)设置发送消息的缓冲区的大小 (totalMemorySize) ,默认是32M。

11)设置消息压缩的类型 (compressionType) 。默认是none表示不压缩。在消息发送的过程中,为了提升发送消息的吞吐量会把消息进行压缩再发送。

12)设置等待元数据更新和缓冲区分配的最长时间 (maxBlockTimeMs) ,默认60S。

13)设置消息投递超时时间 (deliveryTimeoutMs) ,默认120S。消息投递时间是从发送到收到响应的时间。

我们分析了KafkaProducer的核心组件,接下来我们分析下初始化过程中的核心流程。

01

初始化消息累加器

// 初始化消息累加器---缓冲区 this.accumulator = new RecordAccumulator(logContext, config.getInt(ProducerConfig.Batch_SIZE_CONFIG), this.compressionType, 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));

初始化消息累加器对象 「 accumulator 」 ,部分重要参数如下:

1) batchSize : 消息批次大小,默认16KB;

2) compressionType: 消息压缩方式,主要包括none、gzip、snappy、lz4、zstd。默认是不进行压缩,如果你的 Topic 占用的磁盘空间比较多的话,可以考虑启用压缩,以节省资源。

3) lingerMs: 消息 batch 延迟多久再发送的时间,这是吞吐量与延时之间的权衡。为了不频繁发送网络请求,设置延迟时间后 batch 会尽量积累更多的消息再发送出去。

4) retryBackoffMs: 设置失败重试的退避时间。

5) deliveryTimeoutMs: 设置消息投递超时时间。

6) apiVersion: 客户端 api版本。

7) transactionalManager 事务管理器。

8) BufferPool 分配 后续篇在进行深度剖析。

消息累加器---缓冲区的设计是 Kafka Producer 非常优秀和经典的设计 ,Kafka 中消息不是生产后立马就发送给服务端的,而是 会先写入一个缓冲池中,然后直到多条消息组成了一个 Batch,达到一定条件才会一次网络通信把 Batch 发送过去 , 利用该设计来避免 JVM 频繁的 Full GC 的问题 ,后续会单独对其进行深度剖析。

02

初始化集群元数据

元数据的获取涉及的组件比较多,主要分为:

1)KafkaProducer 主线程负责加载元数据 。

2)Sender 子线程负责拉取元数据 。

首先我们来看下 KafkaProducer 主线程是如何加载元数据。

元数据 「 metadata 」 的初始化的时候是在 KafkaProducer 主线程里面的,源代码如下:

// 初始化 Kafka 集群元数据,元数据会保存到客户端中,并与服务端元数据保持一致 if (metadata != null) { this.metadata = metadata; } else { // 初始化集群元数据 this.metadata = new ProducerMetadata(retryBackoffMs, // 元数据过期时间:默认5分钟 config.getLong(ProducerConfig.METADATA_MAX_AGE_CONFIG), // topic最大空闲时间,如果在规定时间没有被访问,将从缓存删除,下次访问时强制获取元数据 config.getLong(ProducerConfig.METADATA_MAX_IDLE_CONFIG), logContext, clusterResourceListeners, Time.SYSTEM); // 启动metadata的引导程序 this.metadata.bootstrap(addresses); }

它会保存在客户端内存中,并与服务端保持准实时的数据一致性, 元数据 主要包含:

1) Kafka 集群节点信息 。

2) Topic 信息 。

3) Topic对应的分区信息

4)ISR列表信息以及分布情况

5)Leader Partition 所在节点

等等

从上面源代码我们可以看出在 KafkaProducer 的构造方法中初始化了元数据类 「 metadata 」 ,然后调用 「 metadata.bootstrap() 」 来启动引导程序,这个时候 metaData 对象里并没有具体的元数据信息,因为客户端还没发送元数据更新的请求 「 获取是通过唤醒 Sender 线程进行发送的 」 。

而具体的发送和拉取,我们将在下一篇中进行剖析。

03

初始化 Sender 线程

// 初始化 Sender 发送线程类,并同时初始化NetworkClient this.sender = newSender(logContext, kafkaClient, this.metadata);

这里非常关键,初始化 「 Sender 」 发送线程类,并同时初始化 「 NetworkClient 」 ,它为 sender 提供了网络IO的能力,后续我们会对其深度剖析。

04

ioThread 启用 Sender 线程

String ioThreadName = NETWORK_THREAD_PREFIX " | " clientId; // 用 ioThread 线程来封装 Sender 线程类,使用 demon 守护线程方式来启动 Sender 线程类 this.ioThread = new KafkaThread(ioThreadName, this.sender, true); this.ioThread.start(); public KafkaThread(final String name, Runnable runnable, boolean daemon) { super(runnable, name); configureThread(name, daemon); } private void configureThread(final String name, boolean daemon) { setDaemon(daemon); setUncaughtExceptionHandler((t, e) -> log.error("Uncaught exception in thread '{}':", name, e)); }

从上面源代码可以看出使用 「 ioThread 」 线程来封装 「 Sender 」 线程类,并使用 demon 守护线程方式来启动 Sender 线程类。

这里的设计模式非常值得我们去学习,就是在设计一些后台线程的时候,可以把 「 线程 本身 」 和 「 线程执行 」 的逻辑分开,Sender 线程就是线程执行的具体逻辑,而 KafkaThread 其实代表了这个 「 线程本身 」 、 「 线程的名字 」 、 「 未捕获的异常处理 」 , 「 deamon 线程设置 」 。 对 Kafka Thread 的启动会自动执行 Sender 线程的 Run() 方法。

05

doSend 发送

用户可以直接使用 「 producer.send() 」 进行数据的发送,先看一下 「 Send() 」 接口的源码实现。

// 向 topic 异步发送数据,此时回调为空 public Future<RecordMetadata> send(ProducerRecord<K, V> record) { return send(record, null); } // 向 topic 异步地发送数据,当发送确认后唤起回调函数 public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) { // intercept the record, which can be potentially modified; this method does not throw exceptions ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record); return doSend(interceptedRecord, callback); }

生产环境我们一般会使用带回调函数的方式去发送,所以最终实现还是调用了 KafkaProducer 的 「 doSend() 」 接口。

该方法只是把消息发送到缓冲区后直接返回,真正的发送是需要等待 Sender 线程把消息从缓冲区将消息取出来后再进行发送。

源码比较长,这里只简单的分析下都做了哪些事情,后续再进行深度剖析,源码如下 :

private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) { TopicPartition tp = null; try { ....省略 // 1.等待元数据更新即确认数据要发送到的 topic 的 metadata 是可用的 clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), nowMs, maxBlockTimeMs); ....省略 // 2.序列化 record的key和value byte[] serializedKey; serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key()); byte[] serializedValue; serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value()); // 3.获取record消息对应的分区 int partition = partition(record, serializedKey, serializedValue, cluster); tp = new TopicPartition(record.topic(), partition); ....省略 // 4.验证消息的大小 ensureValidRecordSize(serializedSize); // 5.组装回调方法和拦截器为一个对象 Callback interceptCallback = new InterceptorCallback<>(callback, this.interceptors, tp); ....省略 // 6.向 accumulator 中追加数据 RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey, serializedValue, headers, interceptCallback, remainingWaitMs, true, nowMs); // 7.新的批次需要重新进行分区 if (result.abortForNewBatch) { int prevPartition = partition; partitioner.onNewBatch(record.topic(), cluster, prevPartition); partition = partition(record, serializedKey, serializedValue, cluster); tp = new TopicPartition(record.topic(), partition); // producer callback will make sure to call both 'callback' and interceptor callback interceptCallback = new InterceptorCallback<>(callback, this.interceptors, tp); result = accumulator.append(tp, timestamp, serializedKey, serializedValue, headers, interceptCallback, remainingWaitMs, false, nowMs); } // 8.如果 batch 已经满了, 则唤醒 sender 线程发送数据 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; } catch (ApiException e) { ....省略 return new FutureFailure(e); } catch (InterruptedException e) { ....省略 throw new InterruptException(e); } catch (KafkaException e) { ....省略 throw e; } catch (Exception e) { ....省略 throw e; } }

06

整体发送流程

kafka 生产数据命令(30分钟带你图解Kafka生产者初始化核心流程)(4)

03 总结

这里,我们一起来总结一下这篇文章的重点。

1、通过 「 场景驱动 」 的方式从生产者调用出发,抛出初始化和发送的过程是怎样的?

2、带你梳理了 「 Kafka Producer 初始化源码全貌 」,包含主线程的 核心组件模块以及消息累加器的初始化、元数据初始化、 Sender 线程初始化流程。

3、最后通过一张整体发送流程图来勾勒出生产者发送消息的全貌。

原文链接https://mp.weixin.qq.com/s?__biz=Mzg3MTcxMDgxNA==&mid=2247495756&idx=1&sn=71d511a88a9784d28b736219a68f93a6&utm_source=tuicool&utm_medium=referral

,

免责声明:本文仅代表文章作者的个人观点,与本站无关。其原创性、真实性以及文中陈述文字和内容未经本站证实,对本文以及其中全部或者部分内容文字的真实性、完整性和原创性本站不作任何保证或承诺,请读者仅作参考,并自行核实相关内容。文章投诉邮箱:anhduc.ph@yahoo.com

    分享
    投诉
    首页