首页  »   网络通信

Kafka-producer端-network层讯息发送的源码探究

网友分享于:2013-08-29  浏览:0次
Kafka-producer端-network层消息发送的源码探究
前面我们对应用层的一些原理进行了特别的源码跟踪分析,今天我们就深入到网络层,看下producer端的sender线程是如何把一个消息发送到server端的。

    Sender实现了Runnable接口,最后被一个IOThread启动,核心的逻辑是在一个void run(long now)方法中去实现的,具体的代码解释之前的博客中有,这里就不一一介绍了,这里种点介绍发送的最后两个流程,即send和poll,具体的代码行为
for (ClientRequest request : requests)
     client.send(request, now);
this.client.poll(pollTimeout, now);

此处的client对象为NetworkClient,实现的是KafkaClient,我们先来看下KafkaClient定义的几个核心的接口
public void send(ClientRequest request, long now);
public List<ClientResponse> poll(long timeout, long now);

send主要是进行ClientRequest的发送的,poll是做具体的IO的
接下来我们看NetworkClient的实现
@Override
public void send(ClientRequest request, long now) {
    String nodeId = request.request().destination();
    if (!canSendRequest(nodeId))
        throw new IllegalStateException("Attempt to send a request to node " + nodeId + " which is not ready.");
    doSend(request, now);
}
private void doSend(ClientRequest request, long now) {
    request.setSendTimeMs(now);
    this.inFlightRequests.add(request);
     selector.send(request.request());
}

send方法主要做了三件事情:1。检查node是否ready;2。把当前请求添加到队列中;3。调用Selector接口的send方法进行发送;此处可以发现,KafkaClient中的发送并不是具体的send实现,具体的send还是依赖的是network层的Selector实现的。
    那我们再来看下Selector的结构,Selector实现了Selectable,Selectable定义的核心方法摘要如下
/**
     * Queue the given request for sending in the subsequent {@link #poll(long) poll()} calls
     * @param send The request to send
     */
    public void send(Send send);

    /**
     * Do I/O. Reads, writes, connection establishment, etc.
     * @param timeout The amount of time to block if there is nothing to do
     * @throws IOException
     */
    public void poll(long timeout) throws IOException;

可以发现定义的方法和NetworkClient的方法是一一对应的,也就是说他们是不同分层中对同一个方法的实现,而且很明显的是KafkaClient是依赖于Selectable的实现的;
那我们再来看下Selectable的实现类Selector,看下对send方法的实现:
public void send(Send send) {
    KafkaChannel channel = channelOrFail(send.destination());
    try {
        channel.setSend(send);
     } catch (CancelledKeyException e) {
        this.failedSends.add(send.destination());
        close(channel);
     }
}

其实就是获取了当前请求的Node的通道-Channel,然后把Channel中的Send设置一个值,并没有实际的IO操作,如建联,读写,释放等,这让我们想到了接口中定义的另外一个方法,没错,就是poll,poll方法才是做实际的IO操作的,和接口中申明的方法一致。
再回到NetworkClient的poll方法中
@Override
    public List<ClientResponse> poll(long timeout, long now) {
        long metadataTimeout = metadataUpdater.maybeUpdate(now);
        try {
            this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
        } 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();
        handleTimedOutRequests(responses, updatedNow);

        // invoke callbacks
        for (ClientResponse response : responses) {
            if (response.request().hasCallback()) {
                try {
                    response.request().callback().onComplete(response);
                } catch (Exception e) {
                    log.error("Uncaught error in request completion:", e);
                }
            }
        }

        return responses;
    }

主要也是做了三件事。1.更新metadata信息;2.调用Selector的send方法进行IO;3.对已经完成的请求进行处理;4. 调用回调函数
看到这里终于看到了metadata相关的实际更新操作,之前在看的时候一只没看到和metadata相关的具体的读写是在哪儿进行的,终于在这里发现了  
接下来我们还是关注Selector的send方法,先上代码:
@Override
    public void poll(long timeout) throws IOException {
        if (timeout < 0)
            throw new IllegalArgumentException("timeout should be >= 0");

        clear();

        if (hasStagedReceives() || !immediatelyConnectedKeys.isEmpty())
            timeout = 0;

        /* check ready keys */
        long startSelect = time.nanoseconds();
        int readyKeys = select(timeout);
        long endSelect = time.nanoseconds();
        currentTimeNanos = endSelect;
        this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds());

        if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
            pollSelectionKeys(this.nioSelector.selectedKeys(), false);
            pollSelectionKeys(immediatelyConnectedKeys, true);
        }

        addToCompletedReceives();

        long endIo = time.nanoseconds();
        this.sensors.ioTime.record(endIo - endSelect, time.milliseconds());
        maybeCloseOldestConnection();
    }


首先获取准备好的事件
int readyKeys = select(timeout);

这里主要调用的NIO的Selector的select方法
如果有准备好的事件
if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
            pollSelectionKeys(this.nioSelector.selectedKeys(), false);
            pollSelectionKeys(immediatelyConnectedKeys, true);
}

在看下具体的pollSelectionKeys方法
private void pollSelectionKeys(Iterable<SelectionKey> selectionKeys, boolean isImmediatelyConnected) {
        Iterator<SelectionKey> iterator = selectionKeys.iterator();
        while (iterator.hasNext()) {
            SelectionKey key = iterator.next();
            iterator.remove();
            KafkaChannel channel = channel(key);

            // register all per-connection metrics at once
            sensors.maybeRegisterConnectionMetrics(channel.id());
            lruConnections.put(channel.id(), currentTimeNanos);

            try {

                /* complete any connections that have finished their handshake (either normally or immediately) */
                if (isImmediatelyConnected || key.isConnectable()) {
                    if (channel.finishConnect()) {
                        this.connected.add(channel.id());
                        this.sensors.connectionCreated.record();
                    } else
                        continue;
                }

                /* if channel is not ready finish prepare */
                if (channel.isConnected() && !channel.ready())
                    channel.prepare();

                /* if channel is ready read from any connections that have readable data */
                if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
                    NetworkReceive networkReceive;
                    while ((networkReceive = channel.read()) != null)
                        addToStagedReceives(channel, networkReceive);
                }

                /* if channel is ready write to any sockets that have space in their buffer and for which we have data */
                if (channel.ready() && key.isWritable()) {
                    Send send = channel.write();
                    if (send != null) {
                        this.completedSends.add(send);
                        this.sensors.recordBytesSent(channel.id(), send.size());
                    }
                }

                /* cancel any defunct sockets */
                if (!key.isValid()) {
                    close(channel);
                    this.disconnected.add(channel.id());
                }

            } catch (Exception e) {
                String desc = channel.socketDescription();
                if (e instanceof IOException)
                    log.debug("Connection with {} disconnected", desc, e);
                else
                    log.warn("Unexpected error from {}; closing connection", desc, e);
                close(channel);
                this.disconnected.add(channel.id());
            }
        }
    }


这里边做的操作就是传统的NIO对SocketChannel的操作模型,获取key类型,然后获取通道,针对不同的key类型做不同的逻辑处理。
但是请注意:此时,我们还没有将数据推到网络中,看代码
/* if channel is ready write to any sockets that have space in their buffer and for which we have data */
if (channel.ready() && key.isWritable()) {
   Send send = channel.write();
   if (send != null) {
       this.completedSends.add(send);
       this.sensors.recordBytesSent(channel.id(), send.size());
   }
}

此时我们只是将数据写入到KafkaChannel中,在看KafkaChannel中的处理
public Send write() throws IOException {
        Send result = null;
        if (send != null && send(send)) {
            result = send;
            send = null;
        }
        return result;
    }

    private boolean send(Send send) throws IOException {
        send.writeTo(transportLayer);
        if (send.completed())
            transportLayer.removeInterestOps(SelectionKey.OP_WRITE);

        return send.completed();
    }


关键看这句
        send.writeTo(transportLayer);

调用send方法把数据写入到TransportLayer层(传输层)
@Override
    public long writeTo(GatheringByteChannel channel) throws IOException {
        long written = channel.write(buffers);
        if (written < 0)
            throw new EOFException("Wrote negative bytes to channel. This shouldn't happen.");
        remaining -= written;
        // This is temporary workaround. As Send , Receive interfaces are being used by BlockingChannel.
        // Once BlockingChannel is removed we can make Send, Receive to work with transportLayer rather than
        // GatheringByteChannel or ScatteringByteChannel.
        if (channel instanceof TransportLayer)
            pending = ((TransportLayer) channel).hasPendingWrites();

        return written;
    }

底层还是调用了Channel的write方法,Kafka实现了自己的GatheringByteChannel,最后调用的还是PlaintextTransportLayer的write方法,看下PlaintextTransportLayer的数据结构,先看下构造方法
public PlaintextTransportLayer(SelectionKey key) throws IOException {
        this.key = key;
        this.socketChannel = (SocketChannel) key.channel();
    }
...
@Override
    public int write(ByteBuffer src) throws IOException {
        return socketChannel.write(src);
    }

可以看到实际调用的就是NIO的SocketChannel的write方法,实现数据从内存到网卡的写入。
至此一个写的分析就基本完成的。
    建连和读取基本类似,这里就不扒代码了。

相关解决方案

最新解决方案