Kafka：Producer生产者发送逻辑 - 源码

1. 简述

KafkaProducer<byte[], byte[]> producer = new KafkaProducer<byte[], byte[]>(props);
producer.send(record, cb);

2. 源码 - KafkaProducer

KafkaProducer

private KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
try {
log.trace("Starting the Kafka producer");
Map<String, Object> userProvidedConfigs = config.originals();
this.producerConfig = config;
this.time = new SystemTime();

// ...
// 省略一些基本的初始化工作
// ...

// records 缓存器
this.accumulator = new RecordAccumulator(config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),
this.totalMemorySize,
this.compressionType,
config.getLong(ProducerConfig.LINGER_MS_CONFIG),
retryBackoffMs,
metrics,
time);

// 更新元数据
List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
this.metadata.update(Cluster.bootstrap(addresses), time.milliseconds());

// 创建客户端
ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config.values());
NetworkClient client = new NetworkClient(
new Selector(config.getLong(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), this.metrics, time, "producer", channelBuilder),
this.metadata,
clientId,
config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION),
config.getLong(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG),
config.getInt(ProducerConfig.SEND_BUFFER_CONFIG),
config.getInt(ProducerConfig.RECEIVE_BUFFER_CONFIG),
this.requestTimeoutMs, time);

// 创建发送线程，daemon形式
// MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = 1时，即 guaranteeMessageOrder
this.sender = new Sender(client,
this.metadata,
this.accumulator,
config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION) == 1,
config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
(short) parseAcks(config.getString(ProducerConfig.ACKS_CONFIG)),
config.getInt(ProducerConfig.RETRIES_CONFIG),
this.metrics,
new SystemTime(),
clientId,
this.requestTimeoutMs);
String ioThreadName = "kafka-producer-network-thread" + (clientId.length() > 0 ? " | " + clientId : "");
this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
this.ioThread.start();

this.errors = this.metrics.sensor("errors");

config.logUnused();
AppInfoParser.registerAppInfo(JMX_PREFIX, clientId);
log.debug("Kafka producer started");
} catch (Throwable t) {
// call close methods if internal objects are already constructed
// this is to prevent resource leak. see KAFKA-2121
close(0, TimeUnit.MILLISECONDS, true);
// now propagate the exception
throw new KafkaException("Failed to construct kafka producer", t);
}
}

3. 源码 - Sender

running

public void run() {
log.debug("Starting Kafka producer I/O thread.");

// main loop, runs until close is called
while (running) {
try {
run(time.milliseconds());
} catch (Exception e) {
log.error("Uncaught error in kafka producer I/O thread: ", e);
}
}

// 调用关闭线程后，执行关闭操作
log.debug("Beginning shutdown of Kafka producer I/O thread, sending remaining records.");

// okay we stopped accepting requests but there may still be
// requests in the accumulator or waiting for acknowledgment,
// wait until these are completed.
// 如果不是强制关闭，且消息累加器accumulator尚有消息未发送，或者客户端client尚有正在处理（in-flight）的请求
while (!forceClose && (this.accumulator.hasUnsent() || this.client.inFlightRequestCount() > 0)) {
try {
// 调用调用待参数的run()方法继续处理
run(time.milliseconds());
} catch (Exception e) {
log.error("Uncaught error in kafka producer I/O thread: ", e);
}
}

// 如果是强制关闭，调用消息累加器accumulator的abortIncompleteBatches()，放弃未处理完的请求
if (forceClose) {
// We need to fail all the incomplete batches and wake up the threads waiting on
// the futures.
this.accumulator.abortIncompleteBatches();
}

// 关闭客户端
try {
this.client.close();
} catch (Exception e) {
log.error("Failed to close network client", e);
}

log.debug("Shutdown of Kafka producer I/O thread has completed.");
}

RecordAccumulator

ready

Metadata

RecordAccumulator

ready

RecordAccumulator

leader

topic partition

RecordBatch

RecordAccumulator

drain

topic partition

RecordAccumulator

muted

RecordAccumulator

drain

topic partition

topic partition

topic partition

RecordAccumulator

drain

ClientRequest

NetworkClient

void run(long now) {
// 获取集群信息
Cluster cluster = metadata.fetch();
// get the list of partitions with data ready to send
// 获取发送就绪 batch 的节点列表
RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);

// 强制更新 metadata
// if there are any partitions whose leaders are not known yet, force metadata update
if (!result.unknownLeaderTopics.isEmpty()) {
// The set of topics with unknown leader contains topics with leader election pending as well as
// topics which may have expired. Add the topic again to metadata to ensure it is included
// and request metadata update, since there are messages to send to the topic.
for (String topic : result.unknownLeaderTopics)
this.metadata.add(topic);
this.metadata.requestUpdate();
}

// 移除未就绪的节点
// remove any nodes we aren't ready to send to
Iterator<Node> iter = result.readyNodes.iterator();
long notReadyTimeout = Long.MAX_VALUE;
while (iter.hasNext()) {
Node node = iter.next();
if (!this.client.ready(node, now)) {
iter.remove();
notReadyTimeout = Math.min(notReadyTimeout, this.client.connectionDelay(node, now));
}
}

// create produce requests
// 创建 producer 请求，遍历就绪节点上的每个partition的第一个batch
// Integer - broker id
// batches - 要发送的batch
Map<Integer, List<RecordBatch>> batches = this.accumulator.drain(cluster,
result.readyNodes,
this.maxRequestSize,
now);

// MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = 1时
// 把要发送的 batches 加入到 RecordAccumulator 的 muted 中
if (guaranteeMessageOrder) {
// Mute all the partitions drained
for (List<RecordBatch> batchList : batches.values()) {
for (RecordBatch batch : batchList)
this.accumulator.mutePartition(batch.topicPartition);
}
}

// 遍历已经超过请求时间的batch，并更新metrics
List<RecordBatch> expiredBatches = this.accumulator.abortExpiredBatches(this.requestTimeout, now);
// update sensors
for (RecordBatch expiredBatch : expiredBatches)
this.sensors.recordErrors(expiredBatch.topicPartition.topic(), expiredBatch.recordCount);

sensors.updateProduceRequestMetrics(batches);

// 对每个节点，创建要发送的请求包
List<ClientRequest> requests = createProduceRequests(batches, now);

// If we have any nodes that are ready to send + have sendable data, poll with 0 timeout so this can immediately
// loop and try sending more data. Otherwise, the timeout is determined by nodes that have partitions with data
// that isn't yet sendable (e.g. lingering, backing off). Note that this specifically does not include nodes
// with sendable data that aren't ready to send since they would cause busy looping.
long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);
if (result.readyNodes.size() > 0) {
log.trace("Nodes with data ready to send: {}", result.readyNodes);
log.trace("Created {} produce requests: {}", requests.size(), requests);
pollTimeout = 0;
}
for (ClientRequest request : requests)
client.send(request, now);

// if some partitions are already ready to be sent, the select time would be 0;
// otherwise if some partition already has some data accumulated but not ready yet,
// the select time will be the time difference between now and its linger expiry time;
// otherwise the select time will be the time difference between now and the metadata expiry time;
this.client.poll(pollTimeout, now);
}

4. 源码 - RecordAccumulator

4.1

ready

函数：

就绪状态 ready

topic partititon

topic partititon

topic partititon

lingerMs

public ReadyCheckResult ready(Cluster cluster, long nowMs) {
Set<Node> readyNodes = new HashSet<>();
long nextReadyCheckDelayMs = Long.MAX_VALUE;
Set<String> unknownLeaderTopics = new HashSet<>();

// 缓冲池是否耗尽
boolean exhausted = this.free.queued() > 0;

// 遍历所有的 TopicPartition 队列
for (Map.Entry<TopicPartition, Deque<RecordBatch>> entry : this.batches.entrySet()) {
TopicPartition part = entry.getKey();
Deque<RecordBatch> deque = entry.getValue();

// 获取该 TopicPartition 的leader broker
Node leader = cluster.leaderFor(part);
synchronized (deque) {
// 若leader未知，则在结果中返回，后续sender会据此更新 Metadata
if (leader == null && !deque.isEmpty()) {
// This is a partition for which leader is not known, but messages are available to send.
// Note that entries are currently not removed from batches when deque is empty.
unknownLeaderTopics.add(part.topic());

// 注意：这里排除了已经放在 muted 中的 TopicPartition, 是为了保证消息顺序（即：MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = 1），也就是若TopicPartition存在正处于发送中的request（未确认），则不属于就绪状态
} else if (!readyNodes.contains(leader) && !muted.contains(part)) {
// 获取第一个Batch，不是移除
RecordBatch batch = deque.peekFirst();
if (batch != null) {
// Batch 是否处于重发策略中，即还在等待下次重发
boolean backingOff = batch.attempts > 0 && batch.lastAttemptMs + retryBackoffMs > nowMs;

// 具体上次重发，已经等待的时间
long waitedTimeMs = nowMs - batch.lastAttemptMs;

// 需要等待的时间：
//    若处于重发策略，等于 配置中的 retry.backoff.ms 时间，默认 100ms
//    否则，等于 配置中的 linger.ms 时间，默认 0 ms
long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;

// 还需要等待的时间
long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);
// Batch 是否已满
boolean full = deque.size() > 1 || batch.records.isFull();
// 是否过时，也就是已经等待时间超过了需要等的时间
boolean expired = waitedTimeMs >= timeToWaitMs;

// 所以最后能够发送的条件为：
//        Batch已满
//    或  已经过时
//    或  缓冲池耗尽
//    或  缓冲池被关闭
//    或  正在要求刷新缓冲池
boolean sendable = full || expired || exhausted || closed || flushInProgress();

// 能够发送 且 不处于重发策略 则为就绪节点
if (sendable && !backingOff) {
readyNodes.add(leader);
} else {
// Note that this results in a conservative estimate since an un-sendable partition may have
// a leader that will later be found to have sendable data. However, this is good enough
// since we'll just wake up and then sleep again for the remaining time.
nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
}
}
}
}
}

return new ReadyCheckResult(readyNodes, nextReadyCheckDelayMs, unknownLeaderTopics);
}

4.2

drain

函数：

topic partition

topic partition

topic partition

public Map<Integer, List<RecordBatch>> drain(Cluster cluster,
Set<Node> nodes,
int maxSize,
long now) {
if (nodes.isEmpty())
return Collections.emptyMap();

Map<Integer, List<RecordBatch>> batches = new HashMap<>();

// 遍历每个节点
for (Node node : nodes) {
int size = 0;

// 该节点下所有作为leader的 partitions
List<PartitionInfo> parts = cluster.partitionsForNode(node.id());
// 该节点下就绪的 batch
List<RecordBatch> ready = new ArrayList<>();
/* to make starvation less likely this loop doesn't start at 0 */
int start = drainIndex = drainIndex % parts.size();

// 遍历所有 partition
do {
PartitionInfo part = parts.get(drainIndex);
TopicPartition tp = new TopicPartition(part.topic(), part.partition());

// Only proceed if the partition has no in-flight batches.
// 只处理不在发送过程中的 topic partition
if (!muted.contains(tp)) {
// 找到缓冲池中该 topic partition 的 队列
Deque<RecordBatch> deque = getDeque(new TopicPartition(part.topic(), part.partition()));
if (deque != null) {
synchronized (deque) {
// 获取队列第一个 Batch
RecordBatch first = deque.peekFirst();
if (first != null) {
boolean backoff = first.attempts > 0 && first.lastAttemptMs + retryBackoffMs > now;
// Only drain the batch if it is not during backoff period.
if (!backoff) {
// 注意，如果加上这个Batch的超过了request最大大小限制，但是该节点下还没有就绪数据，还是会将这个Batch发送出去
// 即若有一个batch大于最大大小，就会将这一个Batch发送出去
if (size + first.records.sizeInBytes() > maxSize && !ready.isEmpty()) {
// there is a rare case that a single batch size is larger than the request size due
// to compression; in this case we will still eventually send this batch in a single
// request
break;
} else {
RecordBatch batch = deque.pollFirst();
batch.records.close();
size += batch.records.sizeInBytes();
ready.add(batch);
batch.drainedMs = now;
}
}
}
}
}
}
this.drainIndex = (this.drainIndex + 1) % parts.size();
} while (start != drainIndex);
batches.put(node.id(), ready);
}
return batches;
}