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

Kafka成长记1:从HelloWorld开始研究Kafka Producer源码原理_第1张图片

成长记不会介绍太多一些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源码原理_第2张图片

这里多说一点,我之前在Zookeeper成长记5提到过源码版本的选择和看源码的方式,这里我就不重复说了。直接将选择后的结果告诉大家,我选择的是kafka-0.10.0.1版本。

所以客户端使用的依赖的GAV(Group-ArtifactId-Version) 是 org.apache.kafka-kafka-clients-0.10.0.1。POM如下所示:


    4.0.0

    org.mfm.learn
    learn-kafka
    0.0.1-SNAPSHOT
    jar

    learn-kafka
    http://maven.apache.org


    
        UTF-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 properties   The producer configs
     */
    public KafkaProducer(Properties properties) {
        this(new ProducerConfig(properties), null, null);
    }

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

我们接着分析,那么接下里就是两条路了,看下重载的构造方法或者是 ProducerConfig是如何解析的。如下:

Kafka成长记1:从HelloWorld开始研究Kafka Producer源码原理_第3张图片

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这个类的源码脉络,看看能不能看出来什么:

Kafka成长记1:从HelloWorld开始研究Kafka Producer源码原理_第4张图片

看着就是有一堆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配置的方法了。如下图所示:

Kafka成长记1:从HelloWorld开始研究Kafka Producer源码原理_第5张图片

那么接下来简单看下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 = 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 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");

就会如下图所示:

Kafka成长记1:从HelloWorld开始研究Kafka Producer源码原理_第6张图片

其实就是解析的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发送出去,必须立即发送出去

......

小结

好了今天我们就先分析到这里,下一节我们继续分析Producer 的创建,通过组件图和流程图的方式看看配置解析之后,执行的重载构造函数又做了那些事情呢?

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

你可能感兴趣的