private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
    .....
    // key序列化
    byte[] serializedKey;
    try {
        serializedKey = keySerializer.serialize(record.topic(), record.headers(), 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", cce);
    }
    // value序列化
    byte[] serializedValue;
    try {
        serializedValue = valueSerializer.serialize(record.topic(), record.headers(), 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", cce);
    }
    ......
}

Kafka内置了几个Serializer，如果需要的话，诸君也可以自定义：
org.apache.kafka.common.serialization.StringSerializer；
org.apache.kafka.common.serialization.LongSerializer；
org.apache.kafka.common.serialization.IntegerSerializer；
org.apache.kafka.common.serialization.ShortSerializer；
org.apache.kafka.common.serialization.FloatSerializer；
org.apache.kafka.common.serialization.DoubleSerializer；
org.apache.kafka.common.serialization.BytesSerializer；
org.apache.kafka.common.serialization.ByteBufferSerializer；
org.apache.kafka.common.serialization.ByteArraySerializer；

1.5 Partition选择

到这里，我们已经有了Topic相关的元数据，但也很快遇到了一个问题：Topic下可能有多个Partition，作为生产者，该将待发消息发给哪个Partition？这就用到了上文提到过的KafkaProducer的一个属性——partitioner。

private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
    ......
    // 确定目标Partition
    int partition = partition(record, serializedKey, serializedValue, cluster);
    ......
}

private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
    // 若ProducerRecord中强制指定了partition, 则以该值为准
    Integer partition = record.partition();
    // 否则调用Partitioner动态计算对应的partition
    return partition != null ?
            partition :
            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) {
    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
    int numPartitions = partitions.size();
    if (keyBytes == null) {
        // 若未指定key
        int nextValue = nextValue(topic);
        List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
        if (availablePartitions.size() > 0) {
            int part = Utils.toPositive(nextValue) % availablePartitions.size();
            return availablePartitions.get(part).partition();
        } else {
            // no partitions are available, give a non-available partition
            return Utils.toPositive(nextValue) % numPartitions;
        }
    } else {
        // hash the keyBytes to choose a partition
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }
}

/**
 * 主线程会调用此方法追加消息
 */
public RecordAppendResult append(TopicPartition tp,
                                 long timestamp,
                                 byte[] key,
                                 byte[] value,
                                 Header[] headers,
                                 Callback callback,
                                 long maxTimeToBlock) throws InterruptedException;
/**
 * Sender线程会调用此方法提取消息 
 */
public Map<Integer, List<ProducerBatch>> drain(Cluster cluster,
                                               Set<Node> nodes,
                                               int maxSize,
                                               long now);

public void send(ClientRequest request, long now);

public List<ClientResponse> poll(long timeout, long now);

/* 实际实现类为 org.apache.kafka.common.network.Selector */
private final Selectable selector;

public void send(ClientRequest request, long now) {
    doSend(request, false, now);
}

private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now) {
    ... ...
    doSend(clientRequest, isInternalRequest, now, builder.build(version));
}

private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now, AbstractRequest request) {
    ... ...
    selector.send(send);
}

public List<ClientResponse> poll(long timeout, long now) {
    ... ...
    this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs));
    ... ...
}

// 在selector.poll前有此行逻辑
long metadataTimeout = metadataUpdater.maybeUpdate(now);
try {            
    this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs));
} catch (IOException e) {
    log.error("Unexpected error during I/O", e);
}

KafkaProducer(ProducerConfig config,
                  Serializer<K> keySerializer,
                  Serializer<V> valueSerializer,
                  Metadata metadata,
                  KafkaClient kafkaClient) {
    ... ...
    this.sender = new Sender(logContext,
            client,
            this.metadata,
            this.accumulator,
            maxInflightRequests == 1,
            config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
            acks,
            retries,
            metricsRegistry.senderMetrics,
            Time.SYSTEM,
            this.requestTimeoutMs,
            config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG),
            this.transactionManager,
            apiVersions);
    String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
    this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
    this.ioThread.start();
    ... ...
}

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

    // 主循环
    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.
    while (!forceClose && (this.accumulator.hasUndrained() || this.client.inFlightRequestCount() > 0)) {
        try {
            run(time.milliseconds());
        } catch (Exception e) {
            log.error("Uncaught error in kafka producer I/O thread: ", e);
        }
    }
    if (forceClose) {
        // We need to fail all the incomplete batches and wake up the threads waiting on
        // the futures.
        log.debug("Aborting incomplete batches due to forced shutdown");
        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.");
}

void run(long now) {
    ... ...
    // 1. 发送请求,并确定下一步的阻塞超时时间
    long pollTimeout = sendProducerData(now);
    // 2. 处理端口事件,poll的timeout为上一步计算结果
    client.poll(pollTimeout, now);
}