成长记不会介绍太多一些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 的创立，通过组件图和流程图的形式看看配置解析之后，执行的重载构造函数又做了那些事件呢？

本文由博客一文多发平台 OpenWrite 公布！