我们都知道Kafka对消息的处理速度非常的快，单机的TPS达到了百万条的数量级。主要是由于Producer端将对个小消息进行合并，进行一个batch message的操作。对于KafkaProducer 的流程设计通过源码的角度进行详细的解析。
这里使用的代码版本为： 0.10.1
- 构造方法： public KafkaProducer(Map<String, Object> configs) {...}
- KafkaProducer 中成员变量
// 如果用户没有配置"client.id"， 则用 "producer-" + PRODUCER_CLIENT_ID_SEQUENCE.getAndIncrement()作为cientId
private static final AtomicInteger PRODUCER_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
// 作为jmx中BeanName 的前缀
private static final String JMX_PREFIX = "kafka.producer";
private String clientId;
//通过此类的`partition`方法将那一条消息负载到指定的topic的partition中
//用户可以自定义扩展此类， 可通过 "partitioner.class"进行配置
private final Partitioner partitioner;
//对message的size做限制， 可通过"max.request.size"进行配置
private final int maxRequestSize;
//作为Produce端申请message内存的大小， 如果下一条消息申请内存时，内存大小不够，则等待。可通过"buffer.memory"进行配置
//具体使用RecordAccumulator中的BufferPool中使用
private final long totalMemorySize;
//作为获取集群信息的元数据类
private final Metadata metadata;
//message消息的累积类， 每次发送消息，都将消息append在RecordAccumulator中
private final RecordAccumulator accumulator;
//消息发送线程，从accumulator中获取可以发送的消息， 进行消息的发送
//放入ioThread线程中， 实例化的时候就会启动
private final Sender sender;
//metrics 数据监控类
private final Metrics metrics;
//作为sender启动类
private final Thread ioThread;
//数据传输的压缩格式
private final CompressionType compressionType;
//消息发送失败的统计类
private final Sensor errors;
private final Time time;
//通过此类将消息的key序列化为传输的byte[]
//用户可自己实现序列化方法， 可通过"key.serializer"进行配置
private final Serializer<K> keySerializer;
//通过此类将消息的value序列化为传输的byte[]
//同样可以自己实现， 可通过"value.serializer"进行配置
private final Serializer<V> valueSerializer;
//作为KafkaProducer实现的的输入参数， 用户配置信息类
private final ProducerConfig producerConfig;
//发送消息时，最大的阻塞时间，the buffer is full or metadata unavailable，可通过"max.block.ms"配置（0.10.1版本）
private final long maxBlockTimeMs;
//发送消息时，发送请求的最大超时时间, 可通过"request.timeout.ms"配置（0.10.1版本）
private final int requestTimeoutMs;
//发送数据的拦截器列表， 对发送ProducerRecord时， 进行一些拦截处理
private final ProducerInterceptors<K, V> interceptors;
- 消息的发送方法 KafkaProducer.send(ProducerRecord<K, V> record, Callback callback)
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
// intercept the record, which can be potentially modified; this method does not throw exceptions
ProducerRecord<K, V> interceptedRecord = this.interceptors == null ? record : this.interceptors.onSend(record);
return doSend(interceptedRecord, callback);
}
private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
TopicPartition tp = null;
try {
//确保metadata对当前record.topic可用，并返回cluster + waitedOnMetadataMs(此方法的阻塞时间)
//可用的条件:metadata中cluster当前topic的partitionsCount ！= null
//1> 用户没有指定partion
//2> 用户指定了partion，必须 partition < partitionsCount (因为partition是从0开始)
ClusterAndWaitTime clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
//计算下面操作最大可阻塞的时间
long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
Cluster cluster = clusterAndWaitTime.cluster;
//将ProducerRecord 中的key跟value根据对应的序列化类序列化为对应的byte[]
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");
}
//如果用户没有指定partition即record.partition != null
//则根据配置的Partitioner进行消息的负载分配
int partition = partition(record, serializedKey, serializedValue, cluster);
//消息序列化的size, 加上消息头的大小 SIZE_LENGTH(INT类型的大小 4) + OFFSET_LENGTH(LONG类型的大小 8)
int serializedSize = Records.LOG_OVERHEAD + Record.recordSize(serializedKey, serializedValue);
//验证单次消息的大小， 必须小于maxRequestSize， 必须小于totalMemorySize
ensureValidRecordSize(serializedSize);
//生成消息发送对应的TopicPartition
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);
//将消息append到accumulator 中
RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey, serializedValue, interceptCallback, remainingWaitMs);
//如果消息满足发送的条件， 则唤醒发送线程， 进行消息的发送
//满足消息发送的条件：
//1> RecordAccumulator中batches对应的
// TopicPartition的消息队列Deque<RecordBatch>的size() > 1;
// 或者当前RecordBatch.isFull()已经满了
// 2> 当前RecordBatch 是新建的， 新建的表示一定有数据
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;
} catch (...) {
// handling exceptions and record the errors;
// for API exceptions return them in the future,
// for other exceptions throw directly
}
}
- 消息的batch方法。 accumulator.append(tp, timestamp, serializedKey, serializedValue, interceptCallback, remainingWaitMs)
/**
* Add a record to the accumulator, return the append result
* <p>
* The append result will contain the future metadata, and flag for whether the appended batch is full or a new batch is created
* <p>
*
* @param tp The topic/partition to which this record is being sent
* @param timestamp The timestamp of the record
* @param key The key for the record
* @param value The value for the record
* @param callback The user-supplied callback to execute when the request is complete
* @param maxTimeToBlock 最大的申请内存的阻塞时间
*/
public RecordAppendResult append(TopicPartition tp,
long timestamp,
byte[] key,
byte[] value,
Callback callback,
long maxTimeToBlock) throws InterruptedException {
//统计， 在append 中的消息的数据信息
appendsInProgress.incrementAndGet();
try {
// 如果batches中存在对应的TopicPartition的消息队列， 直接返回， 否则创建一个
Deque<RecordBatch> dq = getOrCreateDeque(tp);
synchronized (dq) {
if (closed)
throw new IllegalStateException("Cannot send after the producer is closed.");
//将消息放入dq中： 获取dq中最后一个RecordBatch， 如果不存在， 直接返回NULL
// 如果存在， 将消息append到RecordBatch中， 如果RecordBatch没有空间存放，直接返回NULL
// 如果有空间， append进去， 生成一个FutureRecordMetadata，
//==>并通过callback+FutureRecordMetadata实例化一个Thunk, 添加到thunks中， 供消息响应之后回调
RecordAppendResult appendResult = tryAppend(timestamp, key, value, callback, dq);
if (appendResult != null)
return appendResult;
}
//上面流程append不成功， 则重新申请内存，创建Records， 进行append
//申请内存的大小为， batchSize与当前消息需要size的最大值
//free申请内存时： 1： 如果size == poolableSize(即为batchSize)， 从Deque<ByteBuffer> free 队列中获取，
// 如果队列为空，则重新分配一个batchSize的ByteBuffer, 需要跑判断availableMemory是否大于需要分配size
// 如果满足，则直接分配， 否则需要等待内存的内存的释放
// 2: 如果内存不为空，跟1中队列为空之后的分配策略相同
int size = Math.max(this.batchSize, Records.LOG_OVERHEAD + Record.recordSize(key, value));
log.trace("Allocating a new {} byte message buffer for topic {} partition {}", size, tp.topic(), tp.partition());
ByteBuffer buffer = free.allocate(size, maxTimeToBlock);
synchronized (dq) {
// Need to check if producer is closed again after grabbing the dequeue lock.
if (closed)
throw new IllegalStateException("Cannot send after the producer is closed.");
RecordAppendResult appendResult = tryAppend(timestamp, key, value, callback, dq);
if (appendResult != null) {
// Somebody else found us a batch, return the one we waited for! Hopefully this doesn't happen often...
free.deallocate(buffer);
return appendResult;
}
MemoryRecords records = MemoryRecords.emptyRecords(buffer, compression, this.batchSize);
RecordBatch batch = new RecordBatch(tp, records, time.milliseconds());
FutureRecordMetadata future = Utils.notNull(batch.tryAppend(timestamp, key, value, callback, time.milliseconds()));
dq.addLast(batch);
incomplete.add(batch);
return new RecordAppendResult(future, dq.size() > 1 || batch.records.isFull(), true);
}
} finally {
appendsInProgress.decrementAndGet();
}
}
- ioThread 中 Sender 的工作流程
public void run() {
// 主调用流程，循环执行
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.");
// 非强制关闭时， 如果accumulator 跟 client 还有未发送完的消息， 等待发送
while (!forceClose && (this.accumulator.hasUnsent() || this.client.inFlightRequestCount() > 0)) {
try {
run(time.milliseconds());
} catch (Exception e) {
log.error("Uncaught error in kafka producer I/O thread: ", e);
}
}
// 强制关闭， accumulator中数据直接abort
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.");
}
/**
* 方法几点说明：
* 1： guaranteeMessageOrder字段来判断是否需要担保，数据发送的有序性
* kafka这里为了保证消息发送的顺序， 发送一条Record消息， 进行muted操作，响应之后umuted， 就可以继续发送
* 2： 消息重新发送， RecordBatch中字段attempts + lastAttemptMs， attempts>0 表示重新发送的Record,
* 必须满足 batch.lastAttemptMs + retryBackoffMs > nowMs 才能继续发送
* 3： this.client.ready(node, now)
* 必须为连接状态， 即ConnectionState.CONNECTED
* 对于需要权限验证的请求，必须已验证
* InFlightRequests.canSendMore(node)： 当前节点请求队列为空
* 或者队列中第一个请求已完成且queue.size() < this.maxInFlightRequestsPerConnection
*/
void run(long now) {
Cluster cluster = metadata.fetch();
//获取当前accumulator中batches中的数据， readyNodes + nextReadyCheckDelayMs + unknownLeaderTopics
//readyNodes: 同时满足下面两个条件
// 1.可以发送数据， 下面任何一个条件满足即可
// a.数据有满的数据: deque.size() > 1 (一定有一个数据是满的) 或者第一个; 或者deque中第一个数据是满的
// b.数据存放的时间已失效
// c.BufferPool中有等待释放内存的队列有数据
// d.accumulator 中有刷新操作， 此操作是用户进行KafkaProducer.flush()操作
// 2.若是重试数据，已超过重试的阻塞时间，可以重新发送
//nextReadyCheckDelayMs: 对于readyNodes中不满足可以发送数据数据时， 需要等待可以发送数据的时间，即下一个检测准备数据的延迟的时间
//unknownLeaderTopics： batches中的TopicPartition在cluster不能找到Leader且!deque.isEmpty()(有数据需要发送)
RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);
// if there are any partitions whose leaders are not known yet, force metadata update
// 如果返回的数据有不知道Leader的Topic， 则放入metadata 中， 请求更新metadata
if (!result.unknownLeaderTopics.isEmpty()) {
for (String topic : result.unknownLeaderTopics)
this.metadata.add(topic);
this.metadata.requestUpdate();
}
// 对于readyNodes中Node中不能发送数据的直接 移除
// notReadyTimeout 用于this.client.poll(pollTimeout, now);即此方法的最大阻塞时间
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
Map<Integer, List<RecordBatch>> batches = this.accumulator.drain(cluster,
result.readyNodes,
this.maxRequestSize,
now);
if (guaranteeMessageOrder) {
// Mute all the partitions drained
for (List<RecordBatch> batchList : batches.values()) {
for (RecordBatch batch : batchList)
this.accumulator.mutePartition(batch.topicPartition);
}
}
//移除超时的 RecordBatch
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);
}
网友评论