Kafka源码分析－序列3 －Producer －Java NIO(Reactor VS Peactor)

上一篇咱们分析了Metadata的更新机制，其中涉及到一个问题，就是Sender如何跟服务器通讯，也就是网络层。同不少Java项目同样，Kafka client的网络层也是用的Java NIO，而后在上面作了一层封装。

下面首先看一下，在Sender和服务器之间的部分：

能够看到，Kafka client基于Java NIO封装了一个网络层，这个网络层最上层的接口是KakfaClient。其层次关系以下：

在本篇中，先详细对最底层的Java NIO进行讲述。< 喎�"/kf/ware/vc/" target="_blank" class="keylink">vcD4NCjxoMSBpZD0="nio的4大组件">NIO的4大组件

Buffer与Channel

Channel: 在一般的Java网络编程中，咱们知道有一对Socket/ServerSocket对象，每1个socket对象表示一个connection，ServerSocket用于服务器监听新的链接。
在NIO中，与之相对应的一对是SocketChannel/ServerSocketChannel。

下图展现了SocketChannel/ServerSocketChannel的类继承层次

?

1 2 3 4 5 6 7 8 9 10 11 12

public interface Channel extends Closeable {     public boolean isOpen();     public void close() throws IOException; }   public interface ReadableByteChannel extends Channel {     public int read(ByteBuffer dst) throws IOException; }   public interface WritableByteChannel extends Channel {     public int write(ByteBuffer src) throws IOException; }

从代码能够看出，一个Channel最基本的操做就是read/write，而且其传进去的必须是ByteBuffer类型，而不是普通的内存buffer。

Buffer：在NIO中，也有1套围绕Buffer的类继承层次，在此就不详述了。只需知道Buffer就是用来封装channel发送／接收的数据。

Selector

Selector的主要目的是网络事件的 loop 循环，经过调用selector.poll，不断轮询每一个Channel上读写事件

SelectionKey

SelectionKey用来记录一个Channel上的事件集合，每一个Channel对应一个SelectionKey。
SelectionKey也是Selector和Channel之间的关联，经过SelectionKey能够取到对应的Selector和Channel。

关于这4大组件的协做、配合，下面来详细讲述。

4种网络IO模型

epoll与IOCP

在《Unix环境高级编程》中介绍了如下4种IO模型（实际不止4种，但经常使用的就这4种）：

阻塞IO： read/write的时候，阻塞调用

非阻塞IO： read/write，没有数据，立马返回，轮询

IO复用：read/write一次都只能监听一个socket，但对于服务器来说，有成千上完个socket链接，如何用一个函数，能够监听全部的socket上面的读写事件呢？这就是IO复用模型，对应linux上面，就是select/poll/epoll3种技术。

异步IO：linux上没有，windows上对应的是IOCP。

Reactor模式 vs. Preactor模式

相信不少人都据说过网络IO的2种设计模式，关于这2种模式的具体阐述，能够自行google之。

在此处，只想对这2种模式作一个“最通俗的解释“：

Reactor模式：主动模式，所谓主动，是指应用程序不断去轮询，问操做系统，IO是否就绪。Linux下的select/poll/epooll就属于主动模式，须要应用程序中有个循环，一直去poll。
在这种模式下，实际的IO操做仍是应用程序作的。

Proactor模式：被动模式，你把read/write所有交给操做系统，实际的IO操做由操做系统完成，完成以后，再callback你的应用程序。Windows下的IOCP就属于这种模式，再好比C++ Boost中的Asio库，就是典型的Proactor模式。

epoll的编程模型－－3个阶段

在Linux平台上，Java NIO就是基于epoll来实现的。全部基于epoll的框架，都有3个阶段：
注册事件(connect,accept,read,write)， 轮询IO是否就绪，执行实际IO操做。

下面的代码展现了在linux下，用c语言epoll编程的基本框架:

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

//阶段1： 调用epoll_ctl(xx) 注册事件   for( ; ; )     {         nfds = epoll_wait(epfd,events,20,500);     //阶段2：轮询全部的socket           for(i=0;i<nfds;++i) .data.fd="=listenfd)" 0="" accept="" connfd="accept(listenfd,(sockaddr" else="" epollin="" ev.data.fd="connfd;" ev.data.ptr="md;" ev.events="EPOLLIN|EPOLLET;" md="(myepoll_data*)events[i].data.ptr;" n="read(sockfd," sockfd="md-" struct="">fd;                 send( sockfd, md->ptr, strlen((char*)md->ptr), 0 );        //阶段3： 执行实际的io操做                 ev.data.fd=sockfd;                 ev.events=EPOLLIN|EPOLLET;                 epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev); //回到阶段1，从新注册事件             }             else             {                 //其余的处理             }         }     }</nfds;++i)>

一样， NIO中的Selector一样有如下3个阶段，下面把Selector和epoll的使用作个对比：

能够看到，2者只是写法不一样，一样的， 都有这3个阶段。

下面的表格展现了connect, accept, read, write 这4种事件，分别在这3个阶段对应的函数：

下面看一下Kafka client中Selector的核心实现：

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71

@Override public void poll(long timeout) throws IOException {     。。。     clear(); //清空各类状态     if (hasStagedReceives())         timeout = 0;     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) {         Set<selectionkey> keys = this.nioSelector.selectedKeys();         Iterator<selectionkey> iter = keys.iterator();         while (iter.hasNext()) {             SelectionKey key = iter.next();             iter.remove();             KafkaChannel channel = channel(key);               // register all per-connection metrics at once             sensors.maybeRegisterConnectionMetrics(channel.id());             lruConnections.put(channel.id(), currentTimeNanos);               try {                 if (key.isConnectable()) {  //有链接事件                     channel.finishConnect();                     this.connected.add(channel.id());                     this.sensors.connectionCreated.record();                 }                   if (channel.isConnected() && !channel.ready())                     channel.prepare(); //这个只有须要安全检查的SSL需求，普通的不加密的channel，prepare()为空实现                   if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) { //读就绪                     NetworkReceive networkReceive;                     while ((networkReceive = channel.read()) != null)                         addToStagedReceives(channel, networkReceive); //实际的读动做                 }                   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());             }         }     }       addToCompletedReceives();       long endIo = time.nanoseconds();     this.sensors.ioTime.record(endIo - endSelect, time.milliseconds());     maybeCloseOldestConnection(); }</selectionkey></selectionkey>

epoll和selector在注册上的差异

从代码能够看出， Selector和epoll在代码结构上基本同样，但在事件的注册上面有区别：

epoll: 每次read/write以后，都要调用epoll_ctl从新注册

Selector: 注册一次，一直有效，一直会有事件产生，所以须要取消注册。下面来详细分析一下：

connect事件的注册

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

//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.open();         。。。         try {             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);  //构造channel的时候，注册connect事件         KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);         key.attach(channel);         this.channels.put(id, channel);     }

connect事件的取消

?

1 2 3 4 5 6 7 8 9 10 11

//在上面的poll函数中，connect事件就绪，也就是指connect链接完成，链接简历  if (key.isConnectable()) {  //有链接事件        channel.finishConnect();                         ...      }    //PlainTransportLayer  public void finishConnect() throws IOException {         socketChannel.finishConnect();  //调用channel的finishConnect()         key.interestOps(key.interestOps() & ~SelectionKey.OP_CONNECT | SelectionKey.OP_READ); //取消connect事件，新加read事件组册     }

read事件的注册

从上面也能够看出，read事件的注册和connect事件的取消，是同时进行的

read事件的取消

由于read是要一直监听远程，是否有新数据到来，因此不会取消，一直监听

write事件的注册

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

//Selector     public void send(Send send) {         KafkaChannel channel = channelOrFail(send.destination());         try {             channel.setSend(send);         } catch (CancelledKeyException e) {             this.failedSends.add(send.destination());             close(channel);         }     }   //KafkaChannel     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;         this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);  //每调用一次Send，注册一次Write事件     }

Write事件的取消

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

//上面的poll函数里面                     if (channel.ready() && key.isWritable()) { //write事件就绪                         Send send = channel.write(); //在这个write里面，取消了write事件                         if (send != null) {                             this.completedSends.add(send);                             this.sensors.recordBytesSent(channel.id(), send.size());                         }                     }         private boolean send(Send send) throws IOException {         send.writeTo(transportLayer);         if (send.completed())             transportLayer.removeInterestOps(SelectionKey.OP_WRITE);  //取消write事件           return send.completed();     }

总结一下：
（1）“事件就绪“这个概念，对于不一样事件类型，仍是有点歧义的

read事件就绪：这个最好理解，就是远程有新数据到来，须要去read

write事件就绪：这个指什么呢？ 其实指本地的socket缓冲区有没有满。没有满的话，应该就是一直就绪的，可写

connect事件就绪： 指connect链接完成

accept事件就绪：有新的链接进来，调用accept处理

（2）不一样类型事件，处理方式是不同的：

connect事件：注册1次，成功以后，就取消了。有且仅有1次

read事件：注册以后不取消，一直监听

write事件： 每调用一次send，注册1次，send成功，取消注册