2.1.4 选择器处理网络请求

梦里梦外; 2023-01-12 15:53 18阅读 0赞

#### 2.1.4 选择器处理网络请求 ####

生产者客户端会按照节点对消息进行分组，每个节点对应一个客户端请求，那么一个生产者客户端需要管理到多个服务端节点的网络连接。 涉及网络通信时， 一般使用选择器模式 。 选择器使用JavaNIO异步非阻塞方式管理连接和读写请求，它用单个线程就可以管理多个网络连接通道 。 使用选择器的好处是：生产者客户端只需要使用一个选择器，就可以同时和Kafka集群的多个服务端进行网络通信。 为了更好地理解Kafka的网络层通信方式，我们先来复习下Java NIO的一些重要概念。

*  **SocketChannel （客户端网络连接通道）**。底层的字节数据读写都发生在通道上，比如从通道中读取数据、将数据写入通道 。 通道会和字节缓冲区一起使用，从通道中读取数据时需要构造一个缓冲区，调用 channel.read(buffer）就会将通道的数据灌入缓冲区；将数据写入通道时，要先将数据写到缓冲区中，调用 channel. write(buffer）可将缓冲区中的每个字节写入通道 。
 *  **Selector（选择器）** 。 发生在通道上的事件有读和写，选择器会通过选择键的方式监昕读写事件的发生 。
 *  **SelectionKey （选择键）** 。 将通道注册到选择器上， channel. register（ selector）返回选择键，这样就将通道和选择器都关联了起来 。 读写事件发生时，通过选择键可以得到对应的通道，从而进行读写操作。

回顾下前面两节，客户端请求从发送线程经过NetworkClient，最后再到选择器 。 发送线程在运行时分别调用Net队町kCli.ent的连接、发送、轮询方法，而NetworkClient又会调用选择器的连接 、 发送、轮询方法。 下面我们分析这3个方法的具体实现。

1.  客户端连接服务端井建立Kafka~道

选择器的 connect()方法会创建客户端到指定远程服务器的网络连接，连接动作使用 Java的SocketChannel对象完成 。不过，如果直接操作SocketChannl，在选择器的轮询中要做很多工作（比如和字节缓冲区相关的字节复制操作）。 这里创建了更抽象的 Kafka通道（ KafkaChannel ） ， 并使用key.attach(KafkaChannel ） 将选择键和 Kafka 通道关联起来 。 当选择器在轮询时，可以通过key.attachfTlent （） 获取绑定到选择键上的Kafka :ifil道 。 选择器还维护了一个节点编号至U Kafka通道的映射关系，便于客户端根据节点编号获取Kafka:ifil道 。 相关代码如下：

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.open();
            socketChannel.configureBlocking(false);
            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 { 
                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;
            }
            SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT);
            KafkaChannel channel;
            try { 
                channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
            } catch (Exception e) { 
                try { 
                    socketChannel.close();
                } finally { 
                    key.cancel();
                }
                throw new IOException("Channel could not be created for socket " + socketChannel, e);
            }
            key.attach(channel);
            this.channels.put(id, channel);
    
            if (connected) { 
                // OP_CONNECT won't trigger for immediately connected channels
                log.debug("Immediately connected to node {}", channel.id());
                immediatelyConnectedKeys.add(key);
                key.interestOps(0);
            }
        }

SocketChannel 、选择键、传输层、 Kafka通道的关系为 ： SocketChannel注M到选择器上返回选择键，将选择器用于构造传输层，再把传输层用于构造Kafka:ifil道 。 这样Kafka:ifil道就矛1l SocketChannel通过选择键进行了关联，本质上Kafka:ifil道是对原始的SocketChannel的一层包装 。 相关代码如下：

public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize) throws KafkaException { 
            try { 
                PlaintextTransportLayer transportLayer = new PlaintextTransportLayer(key);
                Authenticator authenticator = new DefaultAuthenticator();
                authenticator.configure(transportLayer, this.principalBuilder, this.configs);
                return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize);
            } catch (Exception e) { 
                log.warn("Failed to create channel due to ", e);
                throw new KafkaException(e);
            }
        }

构建Kafka:ifil道的传输层有多种实现，比如纯文材莫式、 sasl 、 ssl加密模式。 PlaintextTransportyer是纯文本的传输层实现。

1.  Kafka通道和网络传输层

网络传输不可避免地需要操作Java I/O的字节缓冲区（ ByteBuffer），传输层则面向底层的字节缓冲区，操作的是字节流。 Kafka通道使用抽象的 Send和NetworkReceive表示网络传输中发送的请求和接收的响应 。 发生在 Kafka通道上的读写操作会利用传输层操作底层字节缓冲区，从而构造出NetworkReceive和 Send对缘。 相关代码如下 ：

public class KafkaChannel { 
        private final String id;
        private final TransportLayer transportLayer;
        private final Authenticator authenticator;
        // Tracks accumulated network thread time. This is updated on the network thread.
        // The values are read and reset after each response is sent.
        private long networkThreadTimeNanos;
        private final int maxReceiveSize;
        private NetworkReceive receive;
        private Send send;
        // Track connection and mute state of channels to enable outstanding requests on channels to be
        // processed after the channel is disconnected.
        private boolean disconnected;
        private boolean muted;
        private ChannelState state;

ByteBufferSend 和lNetwor kRecei.ve都会用字节缓冲 区来缓存通道中 的数据 。 Send 的字节缓冲区表示要发送出去的数据，如果缓冲区的数据都发送完，说明 Send写入完成。 NetworkRecei.ve有两个缓冲区，其中 si.ze缓冲区表示数据的氏度， buffer缓冲区表示数据的内容；如果两个缓冲区都写满了，说明NetworkRecei.ve读取完成 。 相关代码如下 ：

public class ByteBufferSend implements Send { 
    
        private final String destination;
        private final int size;
        protected final ByteBuffer[] buffers;
        private int remaining;
        private boolean pending = false;
    }
    
    public class NetworkReceive implements Receive { 
    
        public final static String UNKNOWN_SOURCE = "";
        public final static int UNLIMITED = -1;
    
        private final String source;
        private final ByteBuffer size;
        private final int maxSize;
        private ByteBuffer buffer;
    }

传输层对SocketChannel做了轻量级的封装，它和SocketChannel一样，两者都实现了操作字节缓冲区的 ScatteringByteChannel和GatheringByteChannel。如图2-10所示，传输层作为Kafka通道的成员变革，当选择器调用 Kafka通道的read()和 write()方法时，最终会通过NetworkReceive.readFrom()和Send.writeTo() 方法调用传输层底层 SocketChannel 的read()和write()方法 。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

1.  Kafka通道上的读写操作

客户端如果要操作Kafka通道，都要通过选择器。 选择器监昕到客户端的读写事件，会获取绑定到选择键上的Kafka通道。Kafka通道会将读写操作交给传输层，传输层再使用最底层的 SocketChannel完成数据传送。如图 2-1 l所示，选择器如果监听到写事件发生，调用write()方法把代表客户端请求的Send对象发送到Kafka通道；选择器如果监听到读事件发生，调用read()方法从 Kafka通道中读取代表服务端响应结果的NetworkReceive。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]  
NetworkClientsend()会调用Selectorsend()，继而调用KafkaChannelsetSend()。客户端发送的每个Send请求，都会被设置到一个Katka:illi道中，如果一个Kafka通道上还有未发送成功的Send请求，则后面的请求就不能发送。即客户端发送请求给服务端，在一个Kafka通道中，一次只能发送一个Send请求。KafkaChannelsetSend()是为wr1te()操作做准备的，setSend会将传人的Send参数设置为Kafka通道的成员变量。KafkaChannelsetSend()也注册了写事件，选择器轮询时监听到写事件，会调用Kafkaζhannelwr1te()方法，将setSend()时保存到KafkaJfil道中的Send发送到传输层的SocketChannel中。

一个Kafka通道一次只能处理一个Send请求，每次Send时都要添加写事件。当Send发送成功后，就要取消写事件。Kafka通道是由事件驱动的，如果没有请求，就不需要监昕写事件，Kafka通道就不需要做写操作。一个完整的发送请求和对应的事件监昕步骤是：设置Send请求至lJKatka通道→注册写事件→发送请求→Send请求发送完成→取消写事件。调用Wr1te()方法的“写操作”只有在完成发送时，才会取消写事件。如果Send在一次wr1te()调用时没有写完，选择键的写事件不会被取消，下次会继续触发wr1te()操作，直到整个Send请求被彻底发送完毕。相关代码如下：

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, connection id is " + id);
            this.send = send;
            this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);
        }
    
        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();
        }

Kafka通道上的读取操作和写入操作类似。读取操作如果一次read()没有完成，也要调用多次read()才能完成。因为读取一次可能只是读取了一丁点，构不成一个完整的NetworkReceive。读取数据时是将通道中的数据读取到NetworkReceiver的缓冲区中，只有缓冲区的数据被填充满，才表示接收到一个完整的NetworkReceive。相关代码如下：

public NetworkReceive read() throws IOException { 
            NetworkReceive result = null;
    
            if (receive == null) { 
                receive = new NetworkReceive(maxReceiveSize, id);
            }
    
            receive(receive);
            if (receive.complete()) { 
                receive.payload().rewind();
                result = receive;
                receive = null;
            }
            return result;
        }

如 图2-1 2所示，选择器轮询到“写事件”，会多次调用 KafkaChannelwrite ()方法发送一个完整的发送请求对象(Send), Kafka通道写人的具体步骤如下 。

1.  发送请求时，通过 Kafka通道的setSend()方法设置要发送 的请求对象，井注册写事件。
2.  客户端轮询到写事件时，会取读取Kafka通道中的发送请求，并发送给 网络通道 。
3.  如果本次写操作没有全部完成，那么由于写事件仍然存在， 客户端还会再次轮询到写事件 。
4.  客户端新的轮询会继续发送请求，如果发送完成，则取消写事件 ，并设置返回结果。
5.  请求发送完成后，加入到completedSends集合中，这个数据会被调用者使用 。
6.  请求已经全部发送完成，重置 send对象为空 ，下一次新的请求才可以继续正常进行。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]

如图2-13所示，选择器轮询到“读事件”，会多次调用 KafkaChannelread ()方法读取一个完整的“网络接收对象”(NetworkRece1ve), Kafka通道读取的具体步骤如下 。

1.  客户端轮询到读事件时，调用 Kafka通道的读方法，如果网络接收对象不存在，则新建一个 。
2.  客户端读取网络通道的数据，并将数据填充到网络连接对象 。
3.  如果本次读操作没有全部完成，客户端还会再次轮询到读事件 。
4.  客户端新的轮询会继续读取网络通道中的数据，如果读取完成，则设置返回结果 。
5.  读取完成后，加入到暂时完成的列表中，这个数据会被调用者使用 。
6.  读取全部完成，重置网络接收对象为空，下一次新的读取请求才可以继续正常进行 。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]

我们已经分析了基于Kafka通道的连接、读取响应、发送请求 ，这些操作的前提条件是必须注册相应的连接 、 读取、写人事件。 然后 ， 选择器在轮询时监昕到有对应的事件发生 ， 会获取选择键对应的Katka:ilfii茧 ， 完成我们前面分析到的各种操作 。

4 . 选择器的轮询

在选择键上处理的读写事件， 分别对应客户端的读取响应和发送请求两个动作 。调用Kafka通道的read()和 wri.te()会得到对应 的 NetworkReceive 和 Send 对象，分别加入 col’lpletedReceives 和completedSends变扯 。 相关代码如下：

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();
            this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds());
    
            if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) { 
                pollSelectionKeys(this.nioSelector.selectedKeys(), false, endSelect);
                pollSelectionKeys(immediatelyConnectedKeys, true, endSelect);
            }
    
            long endIo = time.nanoseconds();
            this.sensors.ioTime.record(endIo - endSelect, time.milliseconds());
    
            // we use the time at the end of select to ensure that we don't close any connections that
            // have just been processed in pollSelectionKeys
            maybeCloseOldestConnection(endSelect);
    
            // Add to completedReceives after closing expired connections to avoid removing
            // channels with completed receives until all staged receives are completed.
            addToCompletedReceives();
        }

写操作会将发送成功的 Send加入collpletedSends，读操作先将读取成功的 NetworkReceive加入stagedReceives，最后全部读完之后 ，才从 stagedReceives复制 至UcollpletedReceives。 collpletedSends和collpletedReceives分别表示在选择器端已经发送完成和接收完成的请求，它们会在NetworkClient调用选择器的轮询启用于不同的 handleCollpleteXXX方法。

选择器的轮询是上面分析的各种基于Kafka通道事件操作的源动力， 在选择器上调用轮询方法，通过不断地注册事件 、 执行事件处理、 取消事件，客户端才会发送请求给服务端，并从服务端读取响应结果。 如 图 2-14所示，选择器轮询检测到的各种事件， 要么提前被注册(比如CONNECT)，要么在处理事件时被注册(比如在finishConnect()中注册READ，在setSend()时注册WRITE)，不同的事件都是交给Kafka通道处理。 Kafka通道的底层网络连接通往的正是远程服务端节点，这样就完成了客户端和服务端的通信。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]

如图2-15所示，不同的注册事件在选择器的轮询下，会触发不同的事件处理。客户端建立连接时注册连接事件(步骤(I))，发送请求时注册写事件(步骤(2))。连接事件的处理合确认成功连接，并注册读事件(步骤3))。只有成功连接后，写事件才会被接着选择到。写事件发生时会将请求发送到服务端，接着客户端就开始等待服务端返回响应结果。由于步骤(3)已经注册了读事件，因此服务端如果返回结果，选择器就能够监听到读事件。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]

现在Java版本的生产者客户端已经分析完毕，表2-1 总结了客户端发送过程涉及的主要组件及其用途 。

<table> 
 <thead> 
  <tr> 
   <th>组件</th> 
   <th>主要用途</th> 
   <th>与上下文组件的关系</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>记录收集器( RecordAccumulator )</td> 
   <td>将消息按照分区存储</td> 
   <td>收集消息，提供消息给发送线程</td> 
  </tr> 
  <tr> 
   <td>发送线程(Sender)</td> 
   <td>针对每个节点都创建一个客户端请求</td> 
   <td>将消息按照节点分组转交给客户端连接管理类</td> 
  </tr> 
  <tr> 
   <td>客户端连接管理类(NetworkClient)</td> 
   <td>管理多个节点的客户端请求</td> 
   <td>驱动选择器工作</td> 
  </tr> 
  <tr> 
   <td>选择器(Selector)</td> 
   <td>以轮询模式驱动不同事件</td> 
   <td>通知网络通道读写操作</td> 
  </tr> 
  <tr> 
   <td>Kafka网络通道(kafkaChannel)</td> 
   <td>负责请求的发送和响应接收</td> 
   <td>从原始网络通道中读写字节缓冲区数据</td> 
  </tr> 
 </tbody> 
</table>

--------------------

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/e8377120edb84f27a6392c3cbf8aa37f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/e7e835b0de694378b168e38e4c9091eb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/6829fef56e0c4f50b980b6d3405a456c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/c01c874ed0484160abee9d9757de4734.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/77a0ea30bd344f6cb32c6b8ac358e268.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/9d6aa06488684aa8abbc7fada4752482.png