Kafka成长记的前4节咱们通过KafkaProducerHelloWorld剖析了Producer配置解析、组件组成、元数据拉取原理。
但KafkaProducerHelloWorld发送音讯的代码并没有剖析完，咱们剖析了如到了如下图所示的地位：
接下来，咱们持续往下剖析，这一节咱们次要剖析下发送音讯的初步序列化和分区路由源码原理。
自定义音讯的初步序列化的形式
在producer.send()执行doSend()的时候，waitOnMetadata拉取元数据胜利之后脉络是什么呢？
private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
TopicPartition tp = null;
try {
// first make sure the metadata for the topic is available
long waitedOnMetadataMs = waitOnMetadata(record.topic(), this.maxBlockTimeMs);
long remainingWaitMs = Math.max(0, this.maxBlockTimeMs - waitedOnMetadataMs);
byte[] serializedKey;
try {
serializedKey = keySerializer.serialize(record.topic(), record.key());
} catch (ClassCastException cce) {
throw new SerializationException("Can't convert key of class " + record.key().getClass().getName() +
" to class " + producerConfig.getClass(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).getName() +
" specified in key.serializer");
}
byte[] serializedValue;
try {
serializedValue = valueSerializer.serialize(record.topic(), record.value());
} catch (ClassCastException cce) {
throw new SerializationException("Can't convert value of class " + record.value().getClass().getName() +
" to class " + producerConfig.getClass(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName() +
" specified in value.serializer");
}
int partition = partition(record, serializedKey, serializedValue, metadata.fetch());
int serializedSize = Records.LOG_OVERHEAD + Record.recordSize(serializedKey, serializedValue);
ensureValidRecordSize(serializedSize);
tp = new TopicPartition(record.topic(), partition);
long timestamp = record.timestamp() == null ? time.milliseconds() : record.timestamp();
log.trace("Sending record {} with callback {} to topic {} partition {}", record, callback, record.topic(), partition);
// producer callback will make sure to call both 'callback' and interceptor callback
Callback interceptCallback = this.interceptors == null ? callback : new InterceptorCallback<>(callback, this.interceptors, tp);
RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey, serializedValue, interceptCallback, remainingWaitMs);
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;
// handling exceptions and record the errors;
// for API exceptions return them in the future,
// for other exceptions throw directly
} catch (ApiException e) {
log.debug("Exception occurred during message send:", e);
if (callback != null)
callback.onCompletion(null, e);
this.errors.record();
if (this.interceptors != null)
this.interceptors.onSendError(record, tp, e);
return new FutureFailure(e);
} catch (Exception e) {
throw e;
}
//省略其余各种异样捕捉
}次要脉络就是：
1）waitOnMetadata 期待元数据拉取
2）keySerializer.serialize和valueSerializer.serialize，很显著就是将Record序列化成byte字节数组
3）通过partition进行路由分区，依照肯定路由策略抉择Topic下的某个分区
4）accumulator.append将音讯放入缓冲器中
5）唤醒Sender线程的selector.select()的阻塞，开始解决内存缓冲器中的数据。
整个脉络如下图：
第二步执行的脉络是应用自定义序列化器，将音讯转换为byte[]数组。咱们就来先看下这块的逻辑。
首先第一个问题就是，自定义的音讯序列化器哪里来的？其实是在配置参数中设置的。还记得KafkaProducerHelloWorld代码么？
// KafkaProducerHelloWorld.java
public static void main(String[] args) throws Exception {
Properties props = new Properties();
props.put("bootstrap.servers", "mengfanmao.org:9092");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "test-key", "test-value");
producer.send(record).get();
Thread.sleep(5 * 1000);
producer.close();
}在之前的KafkaProducerHelloWorld.java中，咱们起初并没有设置序列化参数。后果发消息失败，提醒了如下堆栈：
Exception in thread "main" org.apache.kafka.common.config.ConfigException: Missing required configuration "key.serializer" which has no default value.
at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:421)
at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:55)
at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:62)
at org.apache.kafka.clients.producer.ProducerConfig.<init>(ProducerConfig.java:336)
at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:188)
at org.mfm.learn.kafka.KafkaProducerHelloWorld.main(KafkaProducerHelloWorld.java:20)下面堆栈的信息有没有很相熟？ 提醒的那些类不正是咱们之前钻研配置解析相干的源码类么？ProducerConfig、AbstractConfig、ConfigDef切实是太相熟了。
关上源码ConfigDef，你会发现ConfigDef在解析配置文件时，没有序列化配置会使得new KafkaProducer()这一步间接抛出异样，音讯发送失败。
到这里你是不是能够稍微体验进去，浏览源码的益处之一了？
接着你补充配置下序列化参数如下：
// KafkaProducerHelloWorld.java
public static void main(String[] args) throws Exception {
Properties props = new Properties();
props.put("bootstrap.servers", "mengfanmao.org:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "test-key", "test-value");
producer.send(record).get();
Thread.sleep(5 * 1000);
producer.close();
}音讯发送胜利！咱们补充设置的序列化器是客户端jar包中默认提供的StringSerializer。既然有了音讯序列化器，咱们就来看看它是如何序列化的key和value的。
咱们将之前第二步外围简化，其实就是如下代码：
//KafkaProudcer.java#doSend
ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "test-key", "test-value");
keySerializer.serialize(record.topic(), record.key());
valueSerializer.serialize(record.topic(), record.value());
//StringSerializer.java
public byte[] serialize(String topic, String data) {
try {
if (data == null)
return null;
else
return data.getBytes(encoding);
} catch (UnsupportedEncodingException e) {
throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + encoding);
}
}能够看到StringSerializer的序列化的形式非常简单，就是调用String原始的getBytes()办法而已。(PS:第一个参数居然没有应用…)
序列化真的只是到这里为止了么？必定不是，这个bytes[]数组的数据必定最终须要通过网络发送进来的，这里只是算是初步的一次序列化而已。音讯之后最终的序列化，包含具体的格局，咱们之后钻研Kafka应用原生Java NIO解决粘包和拆包问题时在深入研究。
起码，这里咱们能够失去如下的图了：
音讯基于Topic分区路由源码原理
发送音讯时，拉取到元数据、初步序列化音讯为byte[]数组。之后就是通过元数据信息进行路由，抉择一个Topic对应的Partition发送音讯了。在路由抉择发送音讯的分区时，用到了Metadata中的Cluster元数据，这里带大家回顾下它的构造。
Cluster类的元数据内存构造回顾
List<Node>:Kafka Broker节点,次要是Broker的ip、端口。
Map nodesById，key是broker的id，value是Broker的信息Node
Map partitionsByTopic：每个topic有哪些分区，key是topic名称，value是分区信息列表
Map availablePartitionsByTopic，每个topic有哪些以后可用的分区，key是topic名称，value是分区信息列表
Map partitionsByNode，每个broker上放了哪些分区，key是broker的id，value是分区信息列表
unautorhizedTopics:没有被受权拜访的Topic的列表，如果你的客户端没有被受权拜访某个Topic，音讯队列的权限管制用的很少，这个简直能够疏忽。
你能够断点，看下数据，如下所示：
对集群元数据，你能够发现，依据不同的需要、应用和场景，采纳不同的数据结构来进行寄存，kafka Producer设计了不同的数据结构，其实很多时候咱们是能够学习用相似这种思路写代码的。
回顾了元数据之后，客户端必定能够依据元数据信息进行路由了。那么是如何路由的呢？代码如下：
// KakfaProducer.java
private final Partitioner partitioner;
//#doSend()
int partitionpartition = partition(record, serializedKey, serializedValue, metadata.fetch());
//#partition()
private int partition(ProducerRecord<K, V> record, byte[] serializedKey , byte[] serializedValue, Cluster cluster) {
Integer partition = record.partition();
if (partition != null) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(record.topic());
int lastPartition = partitions.size() - 1;
// they have given us a partition, use it
if (partition < 0 || partition > lastPartition) {
throw new IllegalArgumentException(String.format("Invalid partition given with record: %d is not in the range [0...%d].", partition, lastPartition));
}
return partition;
}
return this.partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue,
cluster);
}这段办法脉络很简略，次要就是依据record是否指定分区partition决定：
1）如果发送的音讯record指定了分区，应用元数据信息Cluster校验后，路由后的分区就是指定的分区编号。
2）如果发送的音讯record没有指定分区，应用一个Partitioner组件partition办法路由决定分区编号。
如下图：
上一节咱们说过ProducerRecord的工夫戳和分区是可选的，默认都是null。也就是说，默认会走到Partitioner组件partition这个分支。
可是问题就来了。Partitioner这个是什么时候初始化的？
因为partitioner这个是KafkaProducer的一个成员变量，你能够搜寻下它。你会发现：
private KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
//省略其余代码...
this.partitioner = config.getConfiguredInstance(ProducerConfig.PARTITIONER_CLASS_CONFIG, Partitioner.class);
//省略其余代码...
}原来是在构造函数时候初始化的，它其实就是通过配置解析失去的。并且有一个默认值DefaultPartitioner。
晓得了这个之后，咱们来看看默认的话是如何路发送的音讯呢？
//DefaultPartitioner.java
private final AtomicInteger counter = new AtomicInteger(new Random().nextInt());
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
if (keyBytes == null) {
int nextValue = counter.getAndIncrement();
List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
if (availablePartitions.size() > 0) {
int part = DefaultPartitioner.toPositive(nextValue) % availablePartitions.size();
return availablePartitions.get(part).partition();
} else {
// no partitions are available, give a non-available partition
return DefaultPartitioner.toPositive(nextValue) % numPartitions;
}
} else {
// hash the keyBytes to choose a partition
return DefaultPartitioner.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
}这个办法脉络次要是：
1）从元数据Cluster中的map获取topic下对应的所有分区和分区数量
2）发送音讯如果没有指定key，则从一个随机数开始，每次通过AtomicInteger递增+1，对分区数量或者可用分区大小取模，取得对应的分区编号
3）发送音讯如果指定key，会对key对应的字节数组执行一个算法murmur2，失去一个int数字，之后对分区数量取模，取得对应分区编号
整个过程如下图所示：
通过下面的路由策略，你能够发现，kafka发送的音讯，哪怕只是指定了topic都是能够的。不须要指定key和partition。不过这样可能会导致音讯乱序。
至于如何保障kafka发送音讯的程序性，除了指定分区和key外，其实还须要其余的配置，比方InFlightRequest的size默认是5，须要设置为1，否则重试的时候也会导致音讯乱序，这些咱们前面会剖析到的。
小结
明天咱们次要摸索了音讯的初步序列化形式、音讯的路由策略。咱们简略小结下：
1）Kafka音讯的初步序列化必须通过配置参数指定，个别应用StringSerializer，不指定会导致发送音讯失败
2）Kafka发送的音讯，Topic必须指定，而Topic下的key和partition可选。
默认的分区路由的策略，反对三种，指定分区，指定分区key，或者不指定分区key
a.同时指定或者只指定partition，因为parttition路由的优先级高于key，会依据指定的parttition编号间接路由音讯。
b.如果只是指定key,会对key对应的字节数组执行一个算法murmur2，失去一个int数字，之后对分区数量取模，取得对应分区编号
c.如果都不指定，则从一个随机数开始，每次通过AtomicInteger递增+1,对分区数量或者可用分区大小取模，取得对应的分区编号
这一节的常识比拟轻松，不晓得大家把握的怎么样了。随着对KafkaProducer的剖析，咱们曾经，缓缓揭开了它神秘的面纱了。前面两节咱们一起来剖析下发送音讯的内存缓冲器的原理，如何分配内存区域，队列机制+batch机制如何将音讯批量发送进来。在之后再剖析下，Kakfa如何解决Java 原生NIO中的拆包和粘包的问题。根本Producer的源码原理就钻研的差不多了。
咱们下一节再见！
本文由博客一文多发平台 OpenWrite 公布！
发表回复