kafka发送者线程一

kafka生产者线程负责生产消息，而将消息发送给broker是有一个专门的发送者线程来处理的，也称之为IO Thread，实现了消息的生产与发送解耦，提高吞吐量。

IO Thread是随着生产者的产生而启动的，在启动发送者线程之前会先初始化一个消息累加器

this.accumulator = new RecordAccumulator(logContext,
                    config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),
                    this.compressionType,
                    lingerMs(config),
                    retryBackoffMs,
                    deliveryTimeoutMs,
                    metrics,
                    PRODUCER_METRIC_GROUP_NAME,
                    time,
                    apiVersions,
                    transactionManager,
                    new BufferPool(this.totalMemorySize, config.getInt(ProducerConfig.BATCH_SIZE_CONFIG), metrics, time, PRODUCER_METRIC_GROUP_NAME));

这里有几个参数需要注意下：
kafka会将发往同一个分区的消息累积到同一个批次中，就会涉及到两个维度，空间和时间，不然会陷入无限等待批次消息的累积
batchSize，该参数就是指的空间，也就是每个批次累积多少消息，或者每个批次的分配的内存大小，太大会造成内存浪费，
太小的话，当消息一多时，会有多个批次需要发送，降低了吞吐量
lingerMs，该参数就是指的时间，就是每个批次延迟时间的上限，但是假如批次信息达到batchSize的值后，lingerMs就会失效
当批次的大小还未达到batchSize时，发送消息会延迟，即使当前kafka没有负载压力的情况下
deliveryTimeoutMs，调send方法到return的时间限制，包含消息的延迟发送时间、收到broker响应的时间、发送失败重试的时间三者总和的时间限制。
该值应该大于或等于request.timeout.ms+linger.ms的时间

当对生产者的相关属性值和累加器都初始化后，kafka会自动启动一个io thread

this.sender = newSender(logContext, kafkaClient, this.metadata);
String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
this.ioThread.start();

其中，KafkaClient是实际负责发送消息的客户端，底层基于nio实现的，与broker进行网络通信；KafkaThread继承了Thread；设置了发送者线程的名称、运行的任务runnable、以及线程模式为daemon，这里为true，因此发送者线程的主要逻辑在于runnable的run方法，也就是Sender的run

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

    // main loop, runs until close is called
    while (running) {
        try {
            runOnce();
        } 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.");

 省略....
}

从run方法可看出，就是不断执行runOnce方法，因此来看下runOnce方法的逻辑

void runOnce() {
    if (transactionManager != null) {
        //忽略....
        //处理事务相关的
    }

    long currentTimeMs = time.milliseconds();
//建立与broker的连接，准备待发送消息的数据
    long pollTimeout = sendProducerData(currentTimeMs);
//处理连接上发生的各种IO事件，包含获取来自broker的数据，发送实际的消息对象
    client.poll(pollTimeout, currentTimeMs);
}

在sendProducerData的方法中，看下以下几个关键的逻辑

Cluster cluster = metadata.fetch();
// 通过累加器获取到准备发送数据的分区
RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);

//和这些分区的节点建立连接，底层是通过nio的方式建立连接的
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.pollDelayMs(node, now));
    }
}
//获取待发送的批次信息
Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(cluster, result.readyNodes, this.maxRequestSize, now);
//省略...

//将批次信息转化为具体的网络请求信息，看如下重载的方法
sendProduceRequests(batches, now);
return pollTimeout;


//简单的说，这个方法就是将与节点连接的channel的监听事件设置为OP_WRITE表示可写的，
//然后构建相应的请求消息体，放置到ByteBuffer中
private void sendProduceRequest(long now, int destination, short acks, int timeout, List<ProducerBatch> batches) {
    if (batches.isEmpty())
        return;

    Map<TopicPartition, MemoryRecords> produceRecordsByPartition = new HashMap<>(batches.size());
    final Map<TopicPartition, ProducerBatch> recordsByPartition = new HashMap<>(batches.size());

  //
    ProduceRequest.Builder requestBuilder = ProduceRequest.Builder.forMagic(minUsedMagic, acks, timeout,
            produceRecordsByPartition, transactionalId);
    RequestCompletionHandler callback = new RequestCompletionHandler() {
        public void onComplete(ClientResponse response) {
            handleProduceResponse(response, recordsByPartition, time.milliseconds());
        }
    };

    String nodeId = Integer.toString(destination);
    ClientRequest clientRequest = client.newClientRequest(nodeId, requestBuilder, now, acks != 0,
            requestTimeoutMs, callback);
    client.send(clientRequest, now);
}

sendProducerData方法就是建立与broker的连接，注册到selector，设置监听事件为op_write，准备好相应的请求数据（待发送的消息）。需要注意的是，此时，消息还没有真正的发送出去。真正发送的消息在runOnce方法内部调用client.poll方法

public List<ClientResponse> poll(long timeout, long now) {
    ensureActive();
   //省略.....
    long metadataTimeout = metadataUpdater.maybeUpdate(now);
    try {
//这里底层就是调用selector的select方法，获取到可以处理IO事件的channel，根据selectionKey的类型进行相对应的处理，
//分别为isConnectable、isReadable、isWritable
        this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs));
    } catch (IOException e) {
        log.error("Unexpected error during I/O", e);
    }
    // process completed actions
    long updatedNow = this.time.milliseconds();
    List<ClientResponse> responses = new ArrayList<>();
    handleCompletedSends(responses, updatedNow);
    handleCompletedReceives(responses, updatedNow);
    handleDisconnections(responses, updatedNow);
    handleConnections();
    handleInitiateApiVersionRequests(updatedNow);
    handleTimedOutRequests(responses, updatedNow);
    completeResponses(responses);

    return responses;
}

下面的两个方法为与broker节点建立连接的底层nio实现

private void initiateConnect(Node node, long now) {
    String nodeConnectionId = node.idString();
    try {
        connectionStates.connecting(nodeConnectionId, now, node.host(), clientDnsLookup);
        InetAddress address = connectionStates.currentAddress(nodeConnectionId);
        log.debug("Initiating connection to node {} using address {}", node, address);
        selector.connect(nodeConnectionId,
                new InetSocketAddress(address, node.port()),
                this.socketSendBuffer,
                this.socketReceiveBuffer);
    } catch (IOException e) {
        //省略
}

public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException {
    ensureNotRegistered(id);
    SocketChannel socketChannel = SocketChannel.open();
    SelectionKey key = null;
    try {
//配置channel为非阻塞模式，并设置channel的发送和接收缓冲区大小
        configureSocketChannel(socketChannel, sendBufferSize, receiveBufferSize);
//建立连接
        boolean connected = doConnect(socketChannel, address);
//将channel注册到selector上
        key = registerChannel(id, socketChannel, SelectionKey.OP_CONNECT);

        if (connected) {
            immediatelyConnectedKeys.add(key);
//若连接成功了，改变channel的监听事件
            key.interestOps(0);
        }
    } catch (IOException | RuntimeException e) {
        if (key != null)
            immediatelyConnectedKeys.remove(key);
        channels.remove(id);
        socketChannel.close();
        throw e;
    }
}

总结，kafka的发送者线程底层使用nio来与broker建立连接与数据通信，因此涉及到如何构造发送消息的ByteBuffer对象，处理来自broker的响应数据等，但是本文只是介绍个大体的方向，并没有对细节进行详情的说明