kafka发送者线程一 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 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 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【kafka发送者线程一】 batches) { if (batches.isEmpty()) return; Map produceRecordsByPartition = new HashMap<>(batches.size()); final Map 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 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 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的响应数据等，但是本文只是介绍个大体的方向，并没有对细节进行详情的说明