Kafka Network层解析

时间 2019-11-09

标签 kafka network 解析栏目 Kafka 繁體版

原文原文链接

咱们知道kafka是基于TCP链接的。其并无像不少中间件使用netty做为TCP服务器。而是本身基于Java NIO写了一套。关于kafka为何没有选用netty的缘由能够看这里。java

对Java NIO不太了解的同窗能够先看下这两篇文章，本文须要读者对NIO有必定的了解。node

segmentfault.com/a/119000001…git

www.jianshu.com/p/0d497fe54…github

更多文章见我的博客：github.com/farmerjohng…apache

几个重要类

先看下Kafka Client的网络层架构，图片来自于这篇文章。segmentfault

本文主要分析的是Network层。服务器

Network层有两个重要的类：Selector和KafkaChannel。网络

这两个类和Java NIO层的java.nio.channels.Selector和Channel有点相似。架构

Selector几个关键字段以下app

// jdk nio中的Selector
java.nio.channels.Selector nioSelector;
// 记录当前Selector的全部链接信息
Map<String, KafkaChannel> channels;
// 已发送完成的请求
List<Send> completedSends;
// 已收到的请求
List<NetworkReceive> completedReceives;
// 尚未彻底收到的请求,对上层不可见
Map<KafkaChannel, Deque<NetworkReceive>> stagedReceives;
// 做为client端，调用connect链接远端时返回true的链接
Set<SelectionKey> immediatelyConnectedKeys;
// 已经完成的链接
List<String> connected;
// 一次读取的最大大小
int maxReceiveSize;

复制代码

从网络层来看kafka是分为client端（producer和consumer，broker做为从时也是client）和server端（broker）的。本文将分析client端是如何创建链接，以及收发数据的。server也是依靠Selector和KafkaChannel进行网络传输。在Network层两端的区别并不大。

创建链接

kafka的client端启动时会调用Selector#connect（下文中如无特殊注明，均指org.apache.kafka.common.network.Selector）方法创建链接。

public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException {
    if (this.channels.containsKey(id))
        throw new IllegalStateException("There is already a connection for id " + id);
    // 建立一个SocketChannel
    SocketChannel socketChannel = SocketChannel.open();
    // 设置为非阻塞模式
    socketChannel.configureBlocking(false);
    // 建立socket并设置相关属性
    Socket socket = socketChannel.socket();
    socket.setKeepAlive(true);
    if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
        socket.setSendBufferSize(sendBufferSize);
    if (receiveBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
        socket.setReceiveBufferSize(receiveBufferSize);
    socket.setTcpNoDelay(true);
    boolean connected;
    try {
        // 调用SocketChannel的connect方法，该方法会向远端发起tcp建连请求
        // 由于是非阻塞的，因此该方法返回时，链接不必定已经创建好（即完成3次握手）。链接若是已经创建好则返回true，不然返回false。通常来讲server和client在一台机器上，该方法可能返回true。
        connected = socketChannel.connect(address);
    } catch (UnresolvedAddressException e) {
        socketChannel.close();
        throw new IOException("Can't resolve address: " + address, e);
    } catch (IOException e) {
        socketChannel.close();
        throw e;
    }
    // 对CONNECT事件进行注册
    SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT);
    KafkaChannel channel;
    try {
        // 构造一个KafkaChannel
        channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
    } catch (Exception e) {
      ...
    }
    // 将kafkachannel绑定到SelectionKey上
    key.attach(channel);
    // 放入到map中，id是远端服务器的名称
    this.channels.put(id, channel);
    // connectct为true表明该链接不会再触发CONNECT事件，因此这里要单独处理
    if (connected) {
        // OP_CONNECT won't trigger for immediately connected channels
        log.debug("Immediately connected to node {}", channel.id());
        // 加入到一个单独的集合中
        immediatelyConnectedKeys.add(key);
        // 取消对该链接的CONNECT事件的监听
        key.interestOps(0);
    }
}
复制代码

这里的流程和标准的NIO流程差很少，须要单独说下的是socketChannel#connect方法返回true的场景，该方法的注释中有提到

* <p> If this channel is in non-blocking mode then an invocation of this
* method initiates a non-blocking connection operation.  If the connection
* is established immediately, as can happen with a local connection, then
* this method returns <tt>true</tt>.  Otherwise this method returns
* <tt>false</tt> and the connection operation must later be completed by
* invoking the {@link #finishConnect finishConnect} method.
复制代码

也就是说在非阻塞模式下，对于local connection，链接可能在立刻就创建好了，那该方法会返回true，对于这种状况，不会再触发以后的connect事件。所以kafka用一个单独的集合immediatelyConnectedKeys将这些特殊的链接记录下来。在接下来的步骤会进行特殊处理。

以后会调用poll方法对网络事件监听：

public void poll(long timeout) throws IOException {
...
// select方法是对java.nio.channels.Selector#select的一个简单封装
int readyKeys = select(timeout);
...
// 若是有就绪的事件或者immediatelyConnectedKeys非空
if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
    // 对已就绪的事件进行处理，第2个参数为false
    pollSelectionKeys(this.nioSelector.selectedKeys(), false, endSelect);
    // 对immediatelyConnectedKeys进行处理。第2个参数为true
    pollSelectionKeys(immediatelyConnectedKeys, true, endSelect);
}

addToCompletedReceives();

...
}

private void pollSelectionKeys(Iterable<SelectionKey> selectionKeys, boolean isImmediatelyConnected, long currentTimeNanos) {
Iterator<SelectionKey> iterator = selectionKeys.iterator();
// 遍历集合
while (iterator.hasNext()) {
    SelectionKey key = iterator.next();
    // 移除当前元素，要否则下次poll又会处理一遍
    iterator.remove();
    // 获得connect时建立的KafkaChannel
    KafkaChannel channel = channel(key);
   ...

    try {
        // 若是当前处理的是immediatelyConnectedKeys集合的元素或处理的是CONNECT事件
        if (isImmediatelyConnected || key.isConnectable()) {
            // finishconnect中会增长READ事件的监听
            if (channel.finishConnect()) {
                this.connected.add(channel.id());
                this.sensors.connectionCreated.record();
                ...
            } else
                continue;
        }

        // 对于ssl的链接还有些额外的步骤
        if (channel.isConnected() && !channel.ready())
            channel.prepare();

        // 若是是READ事件
        if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
            NetworkReceive networkReceive;
            while ((networkReceive = channel.read()) != null)
                addToStagedReceives(channel, networkReceive);
        }

        // 若是是WRITE事件
        if (channel.ready() && key.isWritable()) {
            Send send = channel.write();
            if (send != null) {
                this.completedSends.add(send);
                this.sensors.recordBytesSent(channel.id(), send.size());
            }
        }

        // 若是链接失效
        if (!key.isValid())
            close(channel, true);

    } 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, true);
    } finally {
        maybeRecordTimePerConnection(channel, channelStartTimeNanos);
    }
}
}
复制代码

由于immediatelyConnectedKeys中的链接不会触发CONNNECT事件，因此在poll时会单独对immediatelyConnectedKeys的channel调用finishConnect方法。在明文传输模式下该方法会调用到PlaintextTransportLayer#finishConnect，其实现以下：

public boolean finishConnect() throws IOException {
    // 返回true表明已经链接好了
    boolean connected = socketChannel.finishConnect();
    if (connected)
        // 取消监听CONNECt事件，增长READ事件的监听
        key.interestOps(key.interestOps() & ~SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
    return connected;
}
复制代码

关于immediatelyConnectedKeys更详细的内容能够看看这里。

发送数据

kafka发送数据分为两个步骤：

1.调用Selector#send将要发送的数据保存在对应的KafkaChannel中，该方法并无进行真正的网络IO。

// Selector#send
public void send(Send send) {
    String connectionId = send.destination();
    // 若是所在的链接正在关闭中，则加入到失败集合failedSends中
    if (closingChannels.containsKey(connectionId))
        this.failedSends.add(connectionId);
    else {
        KafkaChannel channel = channelOrFail(connectionId, false);
        try {
            channel.setSend(send);
        } catch (CancelledKeyException e) {
            this.failedSends.add(connectionId);
            close(channel, false);
        }
    }
}

//KafkaChannel#setSend
public void setSend(Send send) {
    // 若是还有数据没有发送出去则报错
    if (this.send != null)
        throw new IllegalStateException("Attempt to begin a send operation with prior send operation still in progress.");
    // 保存下来
    this.send = send;
    // 添加对WRITE事件的监听
    this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);
}
复制代码

调用Selector#poll，在第一步中已经对该channel注册了WRITE事件的监听，因此在当channel可写时,会调用到pollSelectionKeys将数据真正的发送出去。

private void pollSelectionKeys(Iterable<SelectionKey> selectionKeys, boolean isImmediatelyConnected, long currentTimeNanos) {
Iterator<SelectionKey> iterator = selectionKeys.iterator();
// 遍历集合
while (iterator.hasNext()) {
    SelectionKey key = iterator.next();
    // 移除当前元素，要否则下次poll又会处理一遍
    iterator.remove();
    // 获得connect时建立的KafkaChannel
    KafkaChannel channel = channel(key);
   ...

    try {
        ...
 

        // 若是是WRITE事件
        if (channel.ready() && key.isWritable()) {
            // 真正的网络写
            Send send = channel.write();
            // 一个Send对象可能会被拆成几回发送，write非空表明一个send发送完成
            if (send != null) {
                // completedSends表明已发送完成的集合
                this.completedSends.add(send);
                this.sensors.recordBytesSent(channel.id(), send.size());
            }
        }
		...
    } catch (Exception e) {
     ...
    } finally {
        maybeRecordTimePerConnection(channel, channelStartTimeNanos);
    }
}
}
复制代码

当可写时，会调用KafkaChannel#write方法，该方法中会进行真正的网络IO：

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 {
    // 最终调用SocketChannel#write进行真正的写
    send.writeTo(transportLayer);
    if (send.completed())
        // 若是写完了，则移除对WRITE事件的监听
        transportLayer.removeInterestOps(SelectionKey.OP_WRITE);

    return send.completed();
}
复制代码

接收数据

若是远端有发送数据过来，那调用poll方法时，会对接收到的数据进行处理。

public void poll(long timeout) throws IOException {
...
// select方法是对java.nio.channels.Selector#select的一个简单封装
int readyKeys = select(timeout);
...
// 若是有就绪的事件或者immediatelyConnectedKeys非空
if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
    // 对已就绪的事件进行处理，第2个参数为false
    pollSelectionKeys(this.nioSelector.selectedKeys(), false, endSelect);
    // 对immediatelyConnectedKeys进行处理。第2个参数为true
    pollSelectionKeys(immediatelyConnectedKeys, true, endSelect);
}

addToCompletedReceives();

...
}

private void pollSelectionKeys(Iterable<SelectionKey> selectionKeys, boolean isImmediatelyConnected, long currentTimeNanos) {
Iterator<SelectionKey> iterator = selectionKeys.iterator();
// 遍历集合
while (iterator.hasNext()) {
    SelectionKey key = iterator.next();
    // 移除当前元素，要否则下次poll又会处理一遍
    iterator.remove();
    // 获得connect时建立的KafkaChannel
    KafkaChannel channel = channel(key);
   ...

    try {
        ...
 

        // 若是是READ事件
        if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
            NetworkReceive networkReceive;
            // read方法会从网络中读取数据，但可能一次只能读取一个req的部分数据。只有读到一个完整的req的状况下，该方法才返回非null
            while ((networkReceive = channel.read()) != null)
                // 将读到的请求存在stagedReceives中
                addToStagedReceives(channel, networkReceive);
        }
		...
    } catch (Exception e) {
     ...
    } finally {
        maybeRecordTimePerConnection(channel, channelStartTimeNanos);
    }
}
}

private void addToStagedReceives(KafkaChannel channel, NetworkReceive receive) {
    if (!stagedReceives.containsKey(channel))
        stagedReceives.put(channel, new ArrayDeque<NetworkReceive>());

    Deque<NetworkReceive> deque = stagedReceives.get(channel);
    deque.add(receive);
}
复制代码

在以后的addToCompletedReceives方法中会对该集合进行处理。

private void addToCompletedReceives() {
    if (!this.stagedReceives.isEmpty()) {
        Iterator<Map.Entry<KafkaChannel, Deque<NetworkReceive>>> iter = this.stagedReceives.entrySet().iterator();
        while (iter.hasNext()) {
            Map.Entry<KafkaChannel, Deque<NetworkReceive>> entry = iter.next();
            KafkaChannel channel = entry.getKey();
            // 对于client端来讲该isMute返回为false，server端则依靠该方法保证消息的顺序
            if (!channel.isMute()) {
                Deque<NetworkReceive> deque = entry.getValue();
                addToCompletedReceives(channel, deque);
                if (deque.isEmpty())
                    iter.remove();
            }
        }
    }
}
private void addToCompletedReceives(KafkaChannel channel, Deque<NetworkReceive> stagedDeque) {
    // 将每一个channel的第一个NetworkReceive加入到completedReceives
    NetworkReceive networkReceive = stagedDeque.poll();
    this.completedReceives.add(networkReceive);
    this.sensors.recordBytesReceived(channel.id(), networkReceive.payload().limit());
}
复制代码

读出数据后，会先放到stagedReceives集合中，而后在addToCompletedReceives方法中对于每一个channel都会从stagedReceives取出一个NetworkReceive（若是有的话），放入到completedReceives中。

这样作的缘由有两点：

对于SSL的链接来讲，其数据内容是加密的，因此不能精准的肯定本次须要读取的数据大小，只能尽量的多读，这样会致使可能会比请求的数据读的要多。那若是该channel以后没有数据能够读，会致使多读的数据将不会被处理。
kafka须要确保一个channel上request被处理的顺序是其发送的顺序。所以对于每一个channel而言，每次poll上层最多只能看见一个请求，当该请求处理完成以后，再处理其余的请求。在sever端，每次poll后都会将该channel给mute掉，即再也不从该channel上读取数据。当处理完成以后，才将该channelunmute，即以后能够从该socket上读取数据。而client端则是经过InFlightRequests#canSendMore控制。

代码中关于这段逻辑的注释以下：

/* In the "Plaintext" setting, we are using socketChannel to read & write to the network. But for the "SSL" setting, * we encrypt the data before we use socketChannel to write data to the network, and decrypt before we return the responses. * This requires additional buffers to be maintained as we are reading from network, since the data on the wire is encrypted * we won't be able to read exact no.of bytes as kafka protocol requires. We read as many bytes as we can, up to SSLEngine's * application buffer size. This means we might be reading additional bytes than the requested size. * If there is no further data to read from socketChannel selector won't invoke that channel and we've have additional bytes * in the buffer. To overcome this issue we added "stagedReceives" map which contains per-channel deque. When we are * reading a channel we read as many responses as we can and store them into "stagedReceives" and pop one response during * the poll to add the completedReceives. If there are any active channels in the "stagedReceives" we set "timeout" to 0 * and pop response and add to the completedReceives. * Atmost one entry is added to "completedReceives" for a channel in each poll. This is necessary to guarantee that * requests from a channel are processed on the broker in the order they are sent. Since outstanding requests added * by SocketServer to the request queue may be processed by different request handler threads, requests on each * channel must be processed one-at-a-time to guarantee ordering. */
复制代码

End

本文分析了kafka network层的实现，在阅读kafka源码时，若是不把network层搞清楚会比较迷，好比req/resp的顺序保障机制、真正进行网络IO的不是send方法等等。