Kafka成长记1（从HelloWorld开始研究Kafka Producer源码原理） springboot

文章图片

成长记不会介绍太多一些kafka的基础知识，如果有需要的话，之后会有专门的《小白起步营》。成长记的默认大家对kafka的一些概念是熟知的、默认也是会基本Kafka的部署的。当然为了照顾一些小白，第一次涉及的知识我会简单介绍和解释的，熟悉的人就当回顾吧。简单的事情重复做有时也是好事。
Kafka成长记会直接从三个方面开始探索，Producer、Broker、Comsumer。过程中，根据场景会使用之前ZK和JDK成长记介绍源码分析方法。话不多说，让我们直接开始第一节的内容吧！
我们之前研究ZK主要是使用的场景法，找到一些核心入口开始分析的。研究Kafka的源码时候，我们也可以参考之前的方法。不过这次我们不直接从Broker服务端节点入手，先从Producer开始入手研究。会用到一些新的分析源码的思想和方法。
要想分析Kafka Producer的源码原理，首先肯定得有一个入口或者下手的地方。很多人使用Kafka肯定都是从一个Demo开始的。自己部署一台Kafka，之后发送下消息，之后在自己消费一条消息。
KafkaProducerHelloWorld 所以我们就从最简单的一个Kafka Producer的Demo开始，从一个KafkaProducerHelloWorld例子开始Kafka源码原理的探索。
HelloWorld的代码如下：

import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import java.util.Properties; /** * @author fanmao */ public class KafkaProducerHelloWorld {public static void main(String[] args) throws Exception { //配置Kafka的一些参数 Properties props = new Properties(); props.put("bootstrap.servers", "192.168.30.1:9092"); // 创建一个Producer实例 KafkaProducer producer = new KafkaProducer<>(props); // 封装一条消息 ProducerRecord record = new ProducerRecord<>( "test-topic", "test-key", "test-value"); // 同步方式发送消息，会阻塞在这里，直到发送完成 // producer.send(record).get(); // 异步方式发送消息，不阻塞，设置一个监听回调函数即可 producer.send(record, new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if(exception == null) { System.out.println("消息发送成功"); } else { System.out.println("消息发送异常"); } } }); Thread.sleep(5 * 1000); // 退出producer producer.close(); }}

上面的代码例子，虽然非常简单，但是也有自己的脉络。
1）创建KafkaProducer
2）准备消息ProducerRecord
3）发送消息producer.send()
简单画个图：

Kafka成长记1（从HelloWorld开始研究Kafka Producer源码原理）

文章图片

这里多说一点，我之前在Zookeeper成长记5提到过源码版本的选择和看源码的方式，这里我就不重复说了。直接将选择后的结果告诉大家，我选择的是kafka-0.10.0.1版本。
所以客户端使用的依赖的GAV(Group-ArtifactId-Version) 是 org.apache.kafka-kafka-clients-0.10.0.1。POM如下所示：

4.0.0org.mfm.learn learn-kafka 0.0.1-SNAPSHOTjarlearn-kafka http://maven.apache.orgUTF-8 1.8 1.8 org.apache.kafka kafka-clients 0.10.0.1 com.alibaba fastjson 1.2.72

KafkaProducer的创建上面KafkaProducerHelloWorld脉络既然主要分了三步，那我们一步一步来看，首先就是KafkaProducer的创建。我们来一起看看它初始化什么东西？
这里问大家一个问题，这种构造方法的源码原理，一般分析的结果用什么方法会比较好？
没错，组件图或者源码脉络图分析最容易理解了。我们只需要有个大致印象就行。方法有了，一般又会用什么思想呢？连蒙带猜、看看注释，猜测组件的作用，是不是？
好了让我们来试试吧！
new KafkaProducer的代码如下：

/** * A producer is instantiated by providing a set of key-value pairs as configuration. Valid configuration strings * are documented here. * @param propertiesThe producer configs */ public KafkaProducer(Properties properties) { this(new ProducerConfig(properties), null, null); }

构造函数中调用了一个重载的构造函数，我们不着急往下看，先看下注释。大体可以知道，这个构造函数，入参是可以通过Properties设置一些参数，之后肯定是讲这个参数转换成了ProducerConfig对象进行封装，肯定有一定的转换方法。你还记得Zookeeper成长记中是不是也有类似的操作，封装了一个QuorumPeerConfig对象。其实分析多了很多源码，你就逐渐有经验了，更好的能驾轻就熟的分析任何一个源码原理了。这才是我想要让大家学会的，而不是它如何解析，封装成配置对象的。
我们接着分析，那么接下里就是两条路了，看下重载的构造方法或者是 ProducerConfig是如何解析的。如下：

文章图片

Kafka Producer 生产者的配置如何解析的？这一节，我们就先来看看ProducerConfig是如何解析配置文件的。new ProducerConfig()的代码如下：

/* * NOTE: DO NOT CHANGE EITHER CONFIG STRINGS OR THEIR JAVA VARIABLE NAMES AS THESE ARE PART OF THE PUBLIC API AND * CHANGE WILL BREAK USER CODE. * 注意：请勿更改任何配置字符串或它们的JAVA变量名，因为它们是公共API的一部分，更改将破坏用户代码。 */ private static final ConfigDef CONFIG; ProducerConfig(Map props) { super(CONFIG, props); }

这个构造函数的脉络，调用了一个super，竟然有一个父类。看起来比Zookeeper的配置解析封装的多一些，不是简单的一个QuorumPeerConfig。
而且有一个静态变量 ConfigDef CONFIG。你肯定想知道它是个什么东西。
我们可以看下ConfigDef这个类的源码脉络，看看能不能看出来什么：

文章图片

看着就是有一堆define方法、validate方法。关键几个变量，比如一个Map configKeys啥的。好像感觉是放key-value配置的
比如key=bootstrap.servers , value192.168.30.:9092的。
实在猜不到，我们可以再看看ConfigDef这个类的注释。

/** /** * This class is used for specifying the set of expected configurations. For each configuration, you can specify * the name, the type, the default value, the documentation, the group information, the order in the group, * the width of the configuration value and the name suitable for display in the UI. * 此类用于指定期望的配置集。对于每种配置，您可以指定名称，类型，默认值，文档，组信息，组中的顺序，配置值的宽度和适合在UI中显示的名称。 * * You can provide special validation logic used for single configuration validation by overriding {@link Validator}. * 您可以通过覆盖{@link Validator}来提供用于单个配置验证的特殊验证逻辑。 * * Moreover, you can specify the dependents of a configuration. The valid values and visibility of a configuration * may change according to the values of other configurations. You can override {@link Recommender} to get valid * values and set visibility of a configuration given the current configuration values. * 此外，您可以指定配置的从属。配置的有效值和可见性可能会根据其他配置的值而改变。您可以覆盖{@link Recommender}来获得有效值， * 并在给定当前配置值的情况下设置配置的可见性。 * 省略其他... * This class can be used standalone or in combination with {@link AbstractConfig} which provides some additional * functionality for accessing configs. * 此类可以单独使用，也可以与{@link AbstractConfig}结合使用，从而提供一些附加功能访问配置的功能。 */

通过上面的话，你应该就不难看出它的功能了。简单的说就是封装了key-value的配置，可以设置和校验key-value，可以单独使用用于访问配置
知道了这个静态变量的作用后，你点击到ProducerConfig的super，进入父类的构造函数：

public AbstractConfig(ConfigDef definition, Map originals, boolean doLog) { /* check that all the keys are really strings */ for (Object key : originals.keySet()) if (!(key instanceof String)) throw new ConfigException(key.toString(), originals.get(key), "Key must be a string."); this.originals = (Map) originals; this.values = definition.parse(this.originals); this.used = Collections.synchronizedSet(new HashSet()); if (doLog) logAll(); }

上面的代码核心脉络就一句话definition.parse(this.originals); 也就是执行了ConfigDef的parrse方法。
到这里，你想都不用想，这个方法就是转换 Properties为ProducerConfig配置的方法了。如下图所示：

文章图片

那么接下来简单看下parse方法吧，代码如下：

private final Map configKeys = new HashMap<>(); public Map parse(Map props) { // Check all configurations are defined List undefinedConfigKeys = undefinedDependentConfigs(); if (!undefinedConfigKeys.isEmpty()) { String joined = Utils.join(undefinedConfigKeys, ","); throw new ConfigException("Some configurations in are referred in the dependents, but not defined: " + joined); } // parse all known keys Map values = new HashMap<>(); for (ConfigKey key : configKeys.values()) { Object value; // props map contains setting - assign ConfigKey value if (props.containsKey(key.name)) { value = https://www.it610.com/article/parseType(key.name, props.get(key.name), key.type); // props map doesn't contain setting, the key is required because no default value specified - its an error } else if (key.defaultValue =https://www.it610.com/article/= NO_DEFAULT_VALUE) { throw new ConfigException("Missing required configuration \"" + key.name + "\" which has no default value."); } else { // otherwise assign setting its default value value = https://www.it610.com/article/key.defaultValue; } if (key.validator != null) { key.validator.ensureValid(key.name, value); } values.put(key.name, value); } return values; }

这段代码直接看上去有点懵，没关系，还是直接看的核心脉络。
核心脉络是一个for循环，主要遍历了Map configKey这个map，核心逻辑如下：
1）首先通过parseType确认value的类型, 之后根据ConfigKey定义的配置名称，也就是key
2）最后将准备好的key-value配置，放入Map values中返回给了AbstractConfig
这里我们就知道了最终我们配置的Producer参数，就会放入到AbstractConfig的一个Map中,而且Object说明配置的value是区分整数、字符串之类的。比如

Properties props = new Properties(); props.put("bootstrap.servers", "192.168.30.:9092");

就会如下图所示：

文章图片

其实就是解析的Properties的整个过程了。你会发现其实也没有多复杂，就是稍微比Zookeeper封装的复杂点。
不过如果你细心的话，这里就有一个问题了，上面parase方法的for循环，循环的Map configKey 是什么时候初始化的呢？
我们可以倒回去看看。

private static final ConfigDef CONFIG; ProducerConfig(Map props) { super(CONFIG, props); }

还记得调用父类方法前，这个ConfigDef是子类传递给父类的。这个变量又是一个静态的，要想初始化，肯定是有一段静态初始化代码在ProducerConfig中的。你可以找到如下的代码：

/** retries */ public static final String RETRIES_CONFIG = "retries"; private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error." + " Note that this retry is no different than if the client resent the record upon receiving the error." + " Allowing retries without setting " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to 1 will potentially change the" + " ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second" + " succeeds, then the records in the second batch may appear first."; static { CONFIG = new ConfigDef() .define(BOOTSTRAP_SERVERS_CONFIG, Type.LIST, Importance.HIGH, CommonClientConfigs.BOOSTRAP_SERVERS_DOC) .define(BUFFER_MEMORY_CONFIG, Type.LONG, 32 * 1024 * 1024L, atLeast(0L), Importance.HIGH, BUFFER_MEMORY_DOC) .define(RETRIES_CONFIG, Type.INT, 0, between(0, Integer.MAX_VALUE), Importance.HIGH, RETRIES_DOC) .define(ACKS_CONFIG, Type.STRING, "1", in("all", "-1", "0", "1"), Importance.HIGH, ACKS_DOC) .define(COMPRESSION_TYPE_CONFIG, Type.STRING, "none", Importance.HIGH, COMPRESSION_TYPE_DOC) .define(BATCH_SIZE_CONFIG, Type.INT, 16384, atLeast(0), Importance.MEDIUM, BATCH_SIZE_DOC) .define(TIMEOUT_CONFIG, Type.INT, 30 * 1000, atLeast(0), Importance.MEDIUM, TIMEOUT_DOC) .define(LINGER_MS_CONFIG, Type.LONG, 0, atLeast(0L), Importance.MEDIUM, LINGER_MS_DOC) .define(CLIENT_ID_CONFIG, Type.STRING, "", Importance.MEDIUM, CommonClientConfigs.CLIENT_ID_DOC) .define(SEND_BUFFER_CONFIG, Type.INT, 128 * 1024, atLeast(0), Importance.MEDIUM, // 省略其他define .withClientSslSupport() .withClientSaslSupport(); }

这个静态方法的其实就是调用了define方法，初始化了Producer各个配置名称、默认值还有文档说明。最终封装成一个map，value是ConfigKey,初始化了ConfigDef。

private final Map configKeys = new HashMap<>();

public static class ConfigKey { public final String name; public final Type type; public final String documentation; public final Object defaultValue; public final Validator validator; public final Importance importance; public final String group; public final int orderInGroup; public final Width width; public final String displayName; public final List dependents; public final Recommender recommender; }

这个过程虽然没什么，但是重点就来了，默认值。也就说KafkaProducer的配置，默认值都是在这里初始化的。如果你想知道Producer的默认值，就可以看这里了。
这些参数之前公众号的《Kafka入门系列》中都有详细的介绍，我这里介绍了估计你也记不住。之后我们分析源码的时候你在慢慢理解吧。下面我摘录了一些核心配置，供大家回忆下：
Producer核心参数：
metadata.max.age.ms 默认每隔5分钟会刷新下元数据
max.request.size 每个请求的最大大小（1mb）
buffer.memory 缓冲区的内存大小（32mb）
max.block.ms 缓冲区填满之后或元数据拉取最大阻塞时间（60s）
request.timeout.ms 请求超时时间（30s）
batch.size 每个batch的大小默认（16kb）
linger.ms 默认为0，不延迟发送。
可以配置为10ms，10ms内还没有凑成1个batch发送出去，必须立即发送出去
......
小结 【Kafka成长记1（从HelloWorld开始研究Kafka Producer源码原理）】好了今天我们就先分析到这里，下一节我们继续分析Producer 的创建，通过组件图和流程图的方式看看配置解析之后，执行的重载构造函数又做了那些事情呢？