成长记不会介绍太多一些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<String, String> producer = new KafkaProducer<>(props);
// 封装一条消息
ProducerRecord<String, String> 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()
简单画个图：
这里多说一点，我之前在Zookeeper成长记5提到过源码版本的选择和看源码的方式，这里我就不重复说了。直接将选择后的结果告诉大家，我选择的是kafka-0.10.0.1版本。
所以客户端使用的依赖的GAV(Group-ArtifactId-Version) 是 org.apache.kafka-kafka-clients-0.10.0.1。POM如下所示：
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.mfm.learn</groupId>
<artifactId>learn-kafka</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>learn-kafka</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.10.0.1</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.72</version>
</dependency>
</dependencies>
<build>
</build>
</project>
KafkaProducer的创建
上面KafkaProducerHelloWorld脉络既然主要分了三步，那我们一步一步来看，首先就是KafkaProducer的创建。我们来一起看看它初始化什么东西？
这里问大家一个问题，这种构造方法的源码原理，一般分析的结果用什么方法会比较好？
没错，组件图或者源码脉络图分析最容易理解了。我们只需要有个大致印象就行。方法有了，一般又会用什么思想呢？连蒙带猜、看看注释，猜测组件的作用，是不是？
好了让我们来试试吧！
new KafkaProducer的代码如下：
/**
* A producer is instantiated by providing a set of key-value pairs as configuration. Valid configuration strings
* are documented <a href="http://kafka.apache.org/documentation.html#producerconfigs">here</a>.
* @param properties The 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<String, ?>) originals;
this.values = definition.parse(this.originals);
this.used = Collections.synchronizedSet(new HashSet<String>());
if (doLog)
logAll();
}
上面的代码核心脉络就一句话definition.parse(this.originals); 也就是执行了ConfigDef的parrse方法。
到这里，你想都不用想，这个方法就是转换 Properties为ProducerConfig配置的方法了。如下图所示：
那么接下来简单看下parse方法吧，代码如下：
private final Map<String, ConfigKey> configKeys = new HashMap<>();
public Map<String, Object> parse(Map<?, ?> props) {
// Check all configurations are defined
List<String> 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<String, Object> values = new HashMap<>();
for (ConfigKey key : configKeys.values()) {
Object value;
// props map contains setting - assign ConfigKey value
if (props.containsKey(key.name)) {
value = 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 == NO_DEFAULT_VALUE) {
throw new ConfigException("Missing required configuration \"" + key.name + "\" which has no default value.");
} else {
// otherwise assign setting its default value
value = key.defaultValue;
}
if (key.validator != null) {
key.validator.ensureValid(key.name, value);
}
values.put(key.name, value);
}
return values;
}
这段代码直接看上去有点懵，没关系，还是直接看的核心脉络。
核心脉络是一个for循环，主要遍历了Map<String, ConfigKey> configKey这个map，核心逻辑如下：
1）首先通过parseType确认value的类型, 之后根据ConfigKey定义的配置名称，也就是key
2）最后将准备好的key-value配置，放入Map<String, Object> values中返回给了AbstractConfig
这里我们就知道了最终我们配置的Producer参数，**就会放入到AbstractConfig的一个Map<String,Object>中,而且Object说明配置的value是区分整数、字符串之类的。**比如
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.30.:9092");
就会如下图所示：
其实就是解析的Properties的整个过程了。你会发现其实也没有多复杂，就是稍微比Zookeeper封装的复杂点。
不过如果你细心的话，这里就有一个问题了， 上面parase方法的for循环，循环的Map<String, ConfigKey> configKey 是什么时候初始化的呢？
我们可以倒回去看看。
private static final ConfigDef CONFIG;
ProducerConfig(Map<?, ?> props) {
super(CONFIG, props);
}
还记得调用父类方法前，这个ConfigDef是子类传递给父类的。这个变量又是一个静态的，要想初始化，肯定是有一段静态初始化代码在ProducerConfig中的。你可以找到如下的代码：
/** <code>retries</code> */
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 <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> 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<String, ConfigKey> 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<String> 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发送出去，必须立即发送出去
…
小结
好了今天我们就先分析到这里，下一节我们继续分析Producer 的创建，通过组件图和流程图的方式看看配置解析之后，执行的重载构造函数又做了那些事情呢？