我们先通过一张图来了解一下KafkaProducer发送消息的整个流程：
1、ProducerInterceptors对消息进行拦截。
2、Serializer对消息的key和value进行序列化
3、Partitioner为消息选择合适的Partition
4、RecordAccumulator收集消息，实现批量发送
5、Sender从RecordAccumulator获取消息
6、构造ClientRequest
7、将ClientRequest交给NetworkClient，准备发送
8、NetworkClient将请求放入KafkaChannel的缓存
9、执行网络I/O，发送请求
10、收到响应，调用Client Request的回调函数
11、调用RecordBatch的回调函数，最终调用每个消息注册的回调函数
消息发送的过程中，设计两个线程协同工作。主线程首先将业务数据封装成ProducerRecord对象，之后调用send方法将消息放入RecordAccumulator中暂存，Sender线程负责将消息信息构成请求，并最终执行网络I/O的线程，他从Record Accumulator中取出消息并批量发送出去。
接下来我们看一下kafkaProducer的构造函数：
private KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
...
//通过反射机制实例化配置的partitionier类、keySerializer 类、valueSerializer类
this.partitioner = config.getConfiguredInstance(ProducerConfig.PARTITIONER_CLASS_CONFIG, Partitioner.class);
this.keySerializer =config.getConfiguredInstance(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
Serializer.class);
this.valueSerializer =config.getConfiguredInstance(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
Serializer.class);
//创建并更新元数据信息
this.metadata = new Metadata(retryBackoffMs, config.getLong(ProducerConfig.METADATA_MAX_AGE_CONFIG));
List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
this.metadata.update(Cluster.bootstrap(addresses), time.milliseconds());
//创建RecordAccumulator
this.accumulator = new RecordAccumulator(
//指定每个RecordBatch的大小，单位是字节
config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),
this.totalMemorySize,
this.compressionType,
config.getLong(ProducerConfig.LINGER_MS_CONFIG),
retryBackoffMs,
metrics,
time);
//创建NetworkClient，整个是KafkaProducer网络I/O的核心
NetworkClient client = new NetworkClient(
new Selector(config.getLong(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), this.metrics, time, "producer", channelBuilder),
this.metadata,
clientId,
//启动Sender对应的线程
this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
this.ioThread.start();
...
}
生产者对象创建完成，下面我们来看一下send方法：
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
// 首先会对消息进行拦截
ProducerRecord<K, V> interceptedRecord = this.interceptors == null record : this.interceptors.onSend(record);
return doSend(interceptedRecord, callback);
}
调用ProducerInterceptors的onSend 方法对消息进行拦截或者修改
private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
TopicPartition tp = null;
try {
// 会唤醒Send线程更新Metadata中保存的Kafka集群元数据
long waitedOnMetadataMs = waitOnMetadata(record.topic(), this.maxBlockTimeMs);
long remainingWaitMs = Math.max(0, this.maxBlockTimeMs - waitedOnMetadataMs);
byte[] serializedKey;
//序列化key和value
try {
serializedKey = keySerializer.serialize(record.topic(), record.key());
}
...
byte[] serializedValue;
try {
serializedValue = valueSerializer.serialize(record.topic(), record.value());
}
...
//为当前消息选择一个合适的分区
int partition = partition(record, serializedKey, serializedValue, metadata.fetch());
//获取当前消息序列化偶的大小
int serializedSize = Records.LOG_OVERHEAD + Record.recordSize(serializedKey, serializedValue);
//确保当前消息的大小合法
ensureva lidRecordSize(serializedSize);
//根据选择的分区和消息的主题封装成对象
tp = new TopicPartition(record.topic(), partition);
long timestamp = record.timestamp() == null time.milliseconds() : record.timestamp();
// 回调函数
Callback interceptCallback = this.interceptors == null callback : new InterceptorCallback<>(callback, this.interceptors, tp);
//将消息添加到RecordAccumulator中
RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey, serializedValue, interceptCallback, remainingWaitMs);
//唤醒Sender
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;
}
...
}
在这个方法中一共做了下面这几件事：
1、唤醒Sender更新Kafka集群的元数据信息
2、将key和value序列化
3、为当前消息选择一个合适的分区
4、确保消息的大小合法
5、创建要给TopicPartition对象
6、将消息添加到RecordAccumulator中
7、唤醒Sender线程、
对应的时序图就是：
ProducerInterceptors是一个ProducerInterceptor集合，它里面的方法就是循环调用集合里面对象的对应的方法。它可以在消息被发送之前对其进行修改和拦截，也可以先于用户的Callback，对ACK响应进行预处理。
private long waitOnMetadata(String topic, long maxWaitMs) throws InterruptedException {
// 判断元数据里面有没有保存topic，如果没有就添加到topics集合里面
if (!this.metadata.containsTopic(topic))
this.metadata.add(topic);
//如果当前topic的分区的信息不为null，直接返回
if (metadata.fetch().partitionsForTopic(topic) != null)
return 0;
//获取topic的分区信息失败后会进行元数据更新
long begin = time.milliseconds();
//最大等待时间
long remainingWaitMs = maxWaitMs;
//以能否获取到topic分区的详细信息作为判断条件
while (metadata.fetch().partitionsForTopic(topic) == null) {
log.trace("Requesting metadata update for topic {}.", topic);
//获取当前版本号
int version = metadata.requestUpdate();
//唤醒Sender线程
sender.wakeup();
//阻塞等待完成更新
metadata.awaitUpdate(version, remainingWaitMs);
//获取这次等待的时间
long elapsed = time.milliseconds() - begin;
if (elapsed >= maxWaitMs)
throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
if (metadata.fetch().unauthorizedTopics().contains(topic))
throw new TopicAuthorizationException(topic);
remainingWaitMs = maxWaitMs - elapsed;
}
return time.milliseconds() - begin;
}
waitOnMetadata方法里面会根据能否获取到Topic的详细分区为依据判断是否需要进行元数据更新，如果需要，那么就会唤醒Sender线程，阻塞到完成元数据更新，若等待时间超时就会抛出异常。
private int partition(ProducerRecord<K, V> record, byte[] serializedKey , byte[] serializedValue, Cluster cluster) {
//获取指定的分区号
Integer partition = record.partition();
if (partition != null) {
//获取topic的分区的详细信息
List<PartitionInfo> partitions = cluster.partitionsForTopic(record.topic());
int lastPartition = partitions.size() - 1;
//如果我们指定的分区号不合法，那么就会抛出异常
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);
}
再为当前消息选择分区的时候首先会判断用户是否指定了分区号，如果用户指定了分区号并且合法，那么就直接使用这个分区号，如果没有指定就会通过指定的算法选择一个：
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
//首先获取当前topic的所有分区信息的集合
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
//获取分区的数目
int numPartitions = partitions.size();
//如果用户没有指定key
if (keyBytes == null) {
//获取下一个计数的值，为了防止消息都被放到同一个分区
int nextValue = counter.getAndIncrement();
//获取所有有副本的分区信息
List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
//如果存在
if (availablePartitions.size() > 0) {
//将nextValue进行取模运算得到的标号从availablePartitions查找
int part = DefaultPartitioner.toPositive(nextValue) % availablePartitions.size();
return availablePartitions.get(part).partition();
} else {
// 否则使用默认的分区号进行运算
return DefaultPartitioner.toPositive(nextValue) % numPartitions;
}
} else {
// 如果用户指定了key，那么计算方法就是对key进行hash在对分区的数目进行取模
return DefaultPartitioner.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
首先会根据当前消息有没有key来决定是否通过hash算法来指定分区号，如果没有，那么就采用hash算法，如果有，就会先从存在副本的分区里面找一个分区号返回，否则从所有的分区里面返回一个。
那么，KafkaProducer发送消息的大体流程和主线程的工作差不多就介绍完了，下一篇会对存放消息的RecordAccumulator进行介绍。