Netty Channel

Posted Different Java

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Netty Channel相关的知识，希望对你有一定的参考价值。

Channel和Unsafe

Netty提供了自己的Channel和其子类实现，用于异步I/O操作和其他相关操作。

Channel功能说明

Channel工作原理

JDK NIO Channel的缺点：

  1. JDK的SocketChannel和ServerSocketChannel没有统一的Channel接口供业务开发者使用

  2. JDK的SocketChannel和ServerSocketChannel主要职责就是网络I/O操作，由于他们的是SPI类接口，具体的实现由虚拟机厂家决定，所以扩展不方便。

  3. Netty的Channel需要能够跟Netty的整体架构融合在一起，例如I/O模型、基于ChannelPipeline的定制模型。以及基于元数据描述配置化的TCP参数等， 上述JDK的SocketChannel和ServerSocketChannel都没有提供，需要重新封装

Netty Channel设计原理：

  1. 在Channel接口层，采用Facade模式进行统一封装，将网络I/O操作、网络I/O相关联的其他操作封装起来，统一对外提供。

  2. Channel接口的定义尽量大而全，为SocketChannel和ServerSocketChannel提供统一的视图，由不同子类实现不同的功能，公共功能在抽象父类实现， 最大程度上实现功能和接口的重用

  3. 具体实现采用聚合而非包含的方式，将相关的功能类聚合在Channel中，由Channel统一负责分配和调度，功能实现更加灵活。

Channel功能介绍

网络I/O操作

  1. Channel read()：从当前的Channel中读取数据到第一个inbound缓冲区，如果数据被成功读取，触发ChannelHandler.channelRead(ChannelHandler Context context, ChannelPromise promise)事件，读取操作API调用完成之后，紧接着会触发ChannelHandler.channelReadComplete事件

  2. ChannelFuture write(Object msg)：请求将当前的msg通过ChannelPipeline写入到目标Channel中。write操作只是将消息存入到消息发送环形数 组中，并没有真正被发送，只有调用flush操作才会被写入Channel中，发送给对方。

  3. ChannelFuture write(Object msg, ChannelPromise promise)：功能与write(Object msg)相同，但是携带了ChannelPromise参数负责设置 写入操作的结果

  4. ChannelFuture writeAndFlush(Object msg, ChannelPromise promise)：与方法3类似，不同之处在于它会将消息写入到Channel中发送，等价 于单独调用write和flush。

  5. ChannelFuture writeAndFlush(Object msg)：功能等同于4，但是没有携带ChannelPromise参数

  6. Channel flush()：将之前写入消息发送环形数组中的消息全部写入目标Channel中，发送给通信对方。

  7. ChannelFuture close(ChannelPromise promise)：主动关闭当前连接，通过ChannelPromise设置操作结果并进行结果通知，无论操作是否成功， 都可以通过ChannelPromise获取操作结果。该操作会级联触发ChannelPipeline中所有ChannelHandler的ChannelHandler.close(ChannelHandler Context context, ChannelPromise promise)事件

  8. ChannelFuture disconnect(ChannelPromise promise)：请求断开与远程通信对端的连接并使用ChannelPromise来获取操作结果的通知消息。该 方法会级联触发ChannelHandler.disconnect(ChannelHandlerContext context, ChannelPromise promise)事件

  11. ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise)：该功能与9相似，唯一不同的是携带了ChannelPromise 参数用来写入操作结果

  14. ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise)：与13功能类似，携带了ChannelPromise参数来写入操作结果

  15. ChannelConfig config()：获取当前Channel的配置信息，例如CONNECTTIMEOUTMILLIS

  16. boolean isOpen()：判断当前Channel是否已经打开

  17. boolean isRegistered()：判断当前Channel是否已经注册到EventLoop上

  18. boolean isActive()：判断当前Channel是否已经处于激活状态

  19. ChannelMetadata metadata()：获取当前Channel的元数据描述信息，包括TCP参数配置等。

其他API

  1. EventLoop eventLoop()：获取Channel注册的EventLoop。Channel需要注册到EventLoop的多路复用器上，用来处理I/O事件，EventLoop本质上 就是处理网络读写事件的Reactor线程。在Netty中，EventLoop不仅仅用来处理网络事件，也可以用来执行定时任务和用户自定义NioTask等任务。

  2. Channel parent()：对于服务端Channel而言，父Channel为空；对于客户端Channel，它的父Channel就是创建它的ServerSocketChannel

  3. ChannelId id()：返回ChannelId对象，ChannelId是Channel的唯一标识

ChannelId的可能生成策略如下：

  2. 当前的进程ID

  3. 当前系统时间的毫秒——System.currentTimeMillis()

  4. 当前系统时间纳秒数——System.nanoTime()

  5. 32位的随机整形数

  6. 32位自增的序列数

Channel源码

Channel继承关系类图

服务端NioserverSocketChannel继承关系图

客户端NioSocketChannel继承关系图Netty Channel

AbstractChannel源码

成员变量定义

Netty Channel

  1. 首先定义了5个静态全局异常

  2. Channel parent：代表父类Channel

  3. ChannelId id：采用默认方式生成的全局唯一ID

  4. Unsafe unsafe：Unsafe实例

  5. DefaultChannelPipeline pipeline：当前Channel对应的DefaultChannelPipeline

  6. EventLoop eventLoop：当前Channel注册的EventLoop等一系列

通过变量定义可以看出，AbstractChannel聚合了所有Channel使用到的能力对象，由AbstractChannel提供初始化和统一封装，如果功能和子类强相关，则 定义成抽象方法由子类具体实现。

核心API

Netty基于事件驱动，当Channel进行I/O操作时会产生对应的I/O事件，然后驱动事件在ChannelPipeline中传播，由对应的ChannelHandler对事件进行拦截和处理，不关心的事件可以直接忽略。

网络I/O操作直接调用DefaultChannelPipeline的相关方法，由DefaultChannelPipeline中对应的ChannelHandler进行具体的逻辑处理。

AbstractChannel提供了一些公共API的具体实现，例如localAddress()和remoteAddress()，下面看一下remoteAddress的源码：

 
   
   
 
    
  
    
    
  
  1.     @Override
    2. 
  
    
    
  
  2.     public SocketAddress remoteAddress() {
    3. 
  
    
    
  
  3.         SocketAddress remoteAddress = this.remoteAddress;
    4. 
  
    
    
  
  4.         if (remoteAddress == null) {
    5. 
  
    
    
  
  5.             try {
    6. 
  
    
    
  
  6.                 this.remoteAddress = remoteAddress = unsafe().remoteAddress();
    7. 
  
    
    
  
  7.             } catch (Throwable t) {
    8. 
  
    
    
  
  8.                 // Sometimes fails on a closed socket in Windows.
    9. 
  
    
    
  
  9.                 return null;
    10. 
  
    
    
  
  10.             }
    11. 
  
    
    
  
  11.         }
    12. 
  
    
    
  
  12.         return remoteAddress;
    13. 
  
    
    
  
  13.     }
    14.

首先从缓存的成员变量中获取，如果第一次调用为空，需要通过unsafe的remoteAddress获取，它是个抽象方法，具体由对应的Channel子类实现。

AbstractNioChannel源码

成员变量定义

  1. 定义了一个DOCLOSECLOSEDCHANNELEXCEPTION静态全局异常

  2. SelectableChannel ch：由于NIO Channel、NioSocketChannel和NioServerSocketChannel需要公用，所以定义了一个SocketChannel和 ServerSocketChannel的公共父类SelectableChannel，用于设置SelectableChannel参数和进行I/O操作。

  3. int readInterestOp：代表了JDK SelectionKey的OP_READ

  4. volatile SelectionKey selectionKey：该SelectionKey是Channel注册到EventLoop后返回的选择键。由于Channel会面临多个业务线程的并发 写操作，当SelectionKey由SelectionKey修改以后，为了能让其他业务线程感知到变化，所以需要使用volatile保证修改的可见性。

  5. ChannelPromise connectPromise：代表连接操作结果

  6. ScheduledFuture connectTimeoutFuture：连接超时定时器

核心源码

AbstractNioChannel注册源码

 
   
   
 
    
  
    
    
  
  1.     @Override
    2. 
  
    
    
  
  2.     protected void doRegister() throws Exception {
    3. 
  
    
    
  
  3.         boolean selected = false;
    4. 
  
    
    
  
  4.         for (;;) {
    5. 
  
    
    
  
  5.             try {
    6. 
  
    
    
  
  6.                 selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
    7. 
  
    
    
  
  7.                 return;
    8. 
  
    
    
  
  8.             } catch (CancelledKeyException e) {
    9. 
  
    
    
  
  9.                 if (!selected) {
    10. 
  
    
    
  
  10.                     // Force the Selector to select now as the "canceled" SelectionKey may still be
    11. 
  
    
    
  
  11.                     // cached and not removed because no Select.select(..) operation was called yet.
    12. 
  
    
    
  
  12.                     eventLoop().selectNow();
    13. 
  
    
    
  
  13.                     selected = true;
    14. 
  
    
    
  
  14.                 } else {
    15. 
  
    
    
  
  15.                     // We forced a select operation on the selector before but the SelectionKey is still cached
    16. 
  
    
    
  
  16.                     // for whatever reason. JDK bug ?
    17. 
  
    
    
  
  17.                     throw e;
    18. 
  
    
    
  
  18.                 }
    19. 
  
    
    
  
  19.             }
    20. 
  
    
    
  
  20.         }
    21. 
  
    
    
  
  21.     }
    22.

首先定义一个boolean类型的局部变量selected来标识注册操作是否成功，调用SelectableChannel的register方法，将当前的Channel注册到EventLoop 的多路复用器上，SelectableChannel的注册方法定义如下：

 
   
   
 
    
  
    
    
  
  1. public abstract SelectionKey register(Selector sel, int ops, Object att) throws ClosedChannelException;
    2.

注册Channel的时候需要指定监听的网络操作位来表示Channel对哪几类网络事件感兴趣，具体的定义如下：

  1. public static final int OP_READ = 1 << 0： 读操作位

  2. public static final int OP_WRITE = 1 << 2：写操作位

  3. public static final int OP_CONNECT = 1 << 3：客户端连接服务端操作位

  4. public static final int OP_ACCEPT = 1 << 4：服务端接受客户端连接操作位

AbstractNioChannel注册的是0，说明对任何事件不感兴趣，仅仅完成注册操作。注册的时候可以指定附件，后续Channel接收到网络事件通知时可以从SelectionKey中重新获取之前的附件进行处理。如果注册Channel成功，则返回SelectionKey，通过SelectionKey可以从多路复用器中获取Channel对象。

如果当前注册返回的SelectionKey已经被取消，则抛出CancelledKeyException异常，捕获该异常进行处理。如果是第一次处理该异常，调用多路复用器的selectNow()方法将已经取消的selectionKey从多路复用器中删除掉。操作成功之后，将selected置为true，说明之前失效的selectionKey已经被删除掉。继续发起下一次注册操作，如果成功则退出，如果仍然发生CancelledKeyException异常，说明我们无法删除已经被取消的selectionKey，发生这种问题，直接抛出CancelledKeyException异常到上层进行统一处理。

下面看一下准备处理读操作之前需要设置网络操作位为读的代码：

 
   
   
 
    
  
    
    
  
  1.     @Override
    2. 
  
    
    
  
  2.     protected void doBeginRead() throws Exception {
    3. 
  
    
    
  
  3.         // Channel.read() or ChannelHandlerContext.read() was called
    4. 
  
    
    
  
  4.         final SelectionKey selectionKey = this.selectionKey;
    5. 
  
    
    
  
  5.         if (!selectionKey.isValid()) {
    6. 
  
    
    
  
  6.             return;
    7. 
  
    
    
  
  7.         }
    8. 
  
    
    
  
    9. 
  
    
    
  
  9.         readPending = true;
    10. 
  
    
    
  
    11. 
  
    
    
  
  11.         final int interestOps = selectionKey.interestOps();
    12. 
  
    
    
  
  12.         if ((interestOps & readInterestOp) == 0) {
    13. 
  
    
    
  
  13.             selectionKey.interestOps(interestOps | readInterestOp);
    14. 
  
    
    
  
  14.         }
    15. 
  
    
    
  
  15.     }
    16.

获取当前的SelectionKey进行判断，如果可用说明Channel当前状态正常，则可以进行正常的操作位修改。先将等待读设置为true，将SelectionKey当前的操作位与读操作位按位于操作，如果等于0，说明目前并没有设置读操作位，通过interestOps | readInterestOp设置读操作位，最后调用selectionKey的interestOps方法重新设置通道的网络操作位，这样就可以监听网络的读事件。

AbstractNioByteChannel

成员变量定义

private final Runnable flushTask：负责继续写半包消息

API

看下doWrite(ChannelOutboundBuffer in)的源码：

 
   
   
 
    
  
    
    
  
  1.     @Override
    2. 
  
    
    
  
  2.     protected void doWrite(ChannelOutboundBuffer in) throws Exception {
    3. 
  
    
    
  
  3.         int writeSpinCount = config().getWriteSpinCount();
    4. 
  
    
    
  
  4.         do {
    5. 
  
    
    
  
  5.             Object msg = in.current();
    6. 
  
    
    
  
  6.             if (msg == null) {
    7. 
  
    
    
  
  7.                 // Wrote all messages.
    8. 
  
    
    
  
  8.                 clearOpWrite();
    9. 
  
    
    
  
  9.                 // Directly return here so incompleteWrite(...) is not called.
    10. 
  
    
    
  
  10.                 return;
    11. 
  
    
    
  
  11.             }
    12. 
  
    
    
  
  12.             writeSpinCount -= doWriteInternal(in, msg);
    13. 
  
    
    
  
  13.         } while (writeSpinCount > 0);
    14. 
  
    
    
  
    15. 
  
    
    
  
  15.         incompleteWrite(writeSpinCount < 0);
    16. 
  
    
    
  
  16.     }
    17.

首先从消息发送环形数组中弹出一个消息，判断该消息是否为空，如果为空，说明所有的消息都已经发送完成，清除半包标识，退出循环（该循环的次数默认最多16次），设置半包消息有最大处理次数的原因是当循环发送的时候，I/O线程会一直进行写操作，此时I/O线程无法处理其他的I/O操作，例如读新的消息或执行定时任务和NioTask等，如果网络I/O阻塞或者对方接受消息太慢，可能会导致线程假死。看一下清除半包标识clearOpWrite()的逻辑代码：

 
   
   
 
    
  
    
    
  
  1.     protected final void clearOpWrite() {
    2. 
  
    
    
  
  2.         final SelectionKey key = selectionKey();
    3. 
  
    
    
  
  3.         // Check first if the key is still valid as it may be canceled as part of the deregistration
    4. 
  
    
    
  
  4.         // from the EventLoop
    5. 
  
    
    
  
  5.         // See https://github.com/netty/netty/issues/2104
    6. 
  
    
    
  
  6.         if (!key.isValid()) {
    7. 
  
    
    
  
  7.             return;
    8. 
  
    
    
  
  8.         }
    9. 
  
    
    
  
  9.         final int interestOps = key.interestOps();
    10. 
  
    
    
  
  10.         if ((interestOps & SelectionKey.OP_WRITE) != 0) {
    11. 
  
    
    
  
  11.             key.interestOps(interestOps & ~SelectionKey.OP_WRITE);
    12. 
  
    
    
  
  12.         }
    13. 
  
    
    
  
  13.     }
    14.

首先获取当前的SelectionKey，如果当前的SelectionKey已被取消或者无效，直接返回。如果有效，则获取当前的监控的网络操作位，判断当前的网络操作位是否监听写事件，如果正在监听，则取消对写事件的监听。

如果发送的消息不为空，则继续对消息进行处理，源码如下：

 
   
   
 
    
  
    
    
  
  1.     private int doWriteInternal(ChannelOutboundBuffer in, Object msg) throws Exception {
    2. 
  
    
    
  
  2.         if (msg instanceof ByteBuf) {
    3. 
  
    
    
  
  3.             ByteBuf buf = (ByteBuf) msg;
    4. 
  
    
    
  
  4.             if (!buf.isReadable()) {
    5. 
  
    
    
  
  5.                 in.remove();
    6. 
  
    
    
  
  6.                 return 0;
    7. 
  
    
    
  
  7.             }
    8. 
  
    
    
  
    9. 
  
    
    
  
  9.             final int localFlushedAmount = doWriteBytes(buf);
    10. 
  
    
    
  
  10.             if (localFlushedAmount > 0) {
    11. 
  
    
    
  
  11.                 in.progress(localFlushedAmount);
    12. 
  
    
    
  
  12.                 if (!buf.isReadable()) {
    13. 
  
    
    
  
  13.                     in.remove();
    14. 
  
    
    
  
  14.                 }
    15. 
  
    
    
  
  15.                 return 1;
    16. 
  
    
    
  
  16.             }
    17. 
  
    
    
  
  17.         } else if (msg instanceof FileRegion) {
    18. 
  
    
    
  
  18.             FileRegion region = (FileRegion) msg;
    19. 
  
    
    
  
  19.             if (region.transferred() >= region.count()) {
    20. 
  
    
    
  
  20.                 in.remove();
    21. 
  
    
    
  
  21.                 return 0;
    22. 
  
    
    
  
  22.             }
    23. 
  
    
    
  
    24. 
  
    
    
  
  24.             long localFlushedAmount = doWriteFileRegion(region);
    25. 
  
    
    
  
  25.             if (localFlushedAmount > 0) {
    26. 
  
    
    
  
  26.                 in.progress(localFlushedAmount);
    27. 
  
    
    
  
  27.                 if (region.transferred() >= region.count()) {
    28. 
  
    
    
  
  28.                     in.remove();
    29. 
  
    
    
  
  29.                 }
    30. 
  
    
    
  
  30.                 return 1;
    31. 
  
    
    
  
  31.             }
    32. 
  
    
    
  
  32.         } else {
    33. 
  
    
    
  
  33.             // Should not reach here.
    34. 
  
    
    
  
  34.             throw new Error();
    35. 
  
    
    
  
  35.         }
    36. 
  
    
    
  
  36.         return WRITE_STATUS_SNDBUF_FULL;
    37. 
  
    
    
  
  37.     }
    38.

首先判断消息的类型是否是ByteBuf，如果是进行强制转换，判断ByteBuf是否可读，如果不可读，将该消息从消息循环数组中删除，继续循环处理其他消息。如果可读，则由具体的实现子类完成将ByeBuf写入到Channel中，并返回写入的字节数。如果返回的字节数小于等于0，则返回整形的最大数值。如果返回的写入字节数大于0，设置该消息的处理的进度，然后再判断该消息是否可读，如果不可读，就把该消息移除，返回1。

最后还有一块处理半包发送任务的代码incompleteWrite，源码如下：

 
   
   
 
    
  
    
    
  
  1.     protected final void incompleteWrite(boolean setOpWrite) {
    2. 
  
    
    
  
  2.         // Did not write completely.
    3. 
  
    
    
  
  3.         if (setOpWrite) {
    4. 
  
    
    
  
  4.             setOpWrite();
    5. 
  
    
    
  
  5.         } else {
    6. 
  
    
    
  
  6.             // It is possible that we have set the write OP, woken up by NIO because the socket is writable, and then
    7. 
  
    
    
  
  7.             // use our write quantum. In this case we no longer want to set the write OP because the socket is still
    8. 
  
    
    
  
  8.             // writable (as far as we know). We will find out next time we attempt to write if the socket is writable
    9. 
  
    
    
  
  9.             // and set the write OP if necessary.
    10. 
  
    
    
  
  10.             clearOpWrite();
    11. 
  
    
    
  
    12. 
  
    
    
  
  12.             // Schedule flush again later so other tasks can be picked up in the meantime
    13. 
  
    
    
  
  13.             eventLoop().execute(flushTask);
    14. 
  
    
    
  
  14.         }
    15. 
  
    
    
  
  15.     }
    16.

首先判断是否需要设置半包标识，如果需要则调用setOpWrite()来设置半包标识。如果没有设置写操作位，需要启动单独的Runnable flushTask，将其加入到EventLoop中执行，由Runnable负责半包消息的发送，它就是简单的调用flush方法来发送缓冲数组中的消息。

AbstractNioMessageChannel

成员变量定义

boolean inputShutdown

API

 
   
   
 
    
  
    
    
  
  1.     @Override
    2. 
  
    
    
  
  2.     protected void doWrite(ChannelOutboundBuffer in) throws Exception {
    3. 
  
    
    
  
  3.         final SelectionKey key = selectionKey();
    4. 
  
    
    
  
  4.         final int interestOps = key.interestOps();
    5. 
  
    
    
  
    6. 
  
    
    
  
  6.         for (;;) {
    7. 
  
    
    
  
  7.             Object msg = in.current();
    8. 
  
    
    
  
  8.             if (msg == null) {
    9. 
  
    
    
  
  9.                 // Wrote all messages.
    10. 
  
    
    
  
  10.                 if ((interestOps & SelectionKey.OP_WRITE) != 0) {
    11. 
  
    
    
  
  11.                     key.interestOps(interestOps & ~SelectionKey.OP_WRITE);
    12. 
  
    
    
  
  12.                 }
    13. 
  
    
    
  
  13.                 break;
    14. 
  
    
    
  
  14.             }
    15. 
  
    
    
  
  15.             try {
    16. 
  
    
    
  
  16.                 boolean done = false;
    17. 
  
    
    
  
  17.                 for (int i = config().getWriteSpinCount() - 1; i >= 0; i--) {
    18. 
  
    
    
  
  18.                     if (doWriteMessage(msg, in)) {
    19. 
  
    
    
  
  19.                         done = true;
    20. 
  
    
    
  
  20.                         break;
    21. 
  
    
    
  
  21.                     }
    22. 
  
    
    
  
  22.                 }
    23. 
  
    
    
  
    24. 
  
    
    
  
  24.                 if (done) {
    25. 
  
    
    
  
  25.                     in.remove();
    26. 
  
    
    
  
  26.                 } else {
    27. 
  
    
    
  
  27.                     // Did not write all messages.
    28. 
  
    
    
  
  28.                     if ((interestOps & SelectionKey.OP_WRITE) == 0) {
    29. 
  
    
    
  
  29.                         key.interestOps(interestOps | SelectionKey.OP_WRITE);
    30. 
  
    
    
  
  30.                     }
    31. 
  
    
    
  
  31.                     break;
    32. 
  
    
    
  
  32.                 }
    33. 
  
    
    
  
  33.             } catch (Exception e) {
    34. 
  
    
    
  
  34.                 if (continueOnWriteError()) {
    35. 
  
    
    
  
  35.                     in.remove(e);
    36. 
  
    
    
  
  36.                 } else {
    37. 
  
    
    
  
  37.                     throw e;
    38. 
  
    
    
  
  38.                 }
    39. 
  
    
    
  
  39.             }
    40. 
  
    
    
  
  40.         }
    41. 
  
    
    
  
  41.     }
    42.

在循环体中对消息进行发送，首先从ChannelOutboundBuffer中弹出一条消息进行处理，如果消息为空，说明发送缓冲区为空，所有消息都被发送完成。此时清除写半包标识，退出循环。然后借用writeSpinCount对单条消息进行发送，调用doWriteMessage判断消息是否发送成功，如果成功，则将发送标识done设置为true，退出循环，否则继续执行循环，知道执行writeSpinCount次。发送完成后，判断发送结果，如果当前的消息被完全发送出去，则将该消息从缓冲数组中删除；否则设置半包标识，注册SelectionKey.OP_WRITE到多路复用器上，由多路复用器轮询对应的Channel重新发送尚未发送完全的半包消息。

AbstractNioMessageChannel和AbstractNioByteChannel不同之处是前者发送的是POJO对象，后者发送的是ByteBuf或者FileRegion。

NioServerSocketChannel

成员变量和静态方法

  1. ChannelMetadata METADATA：Channel元数据信息

  2. ServerSocketChannelConfig config：用于配置ServerSocketChannel的TCP参数

  3. ServerSocketChannel newSocket(SelectorProvider provider)：借助SelectorProvider的openServerSocketChannel方法打开新的ServerSocketChannel

API

 
   
   
 
    
  
    
    
  
  1.     @Override
    2. 
  
    
    
  
  2.     public boolean isActive() {
    3. 
  
    
    
  
  3.         return javaChannel().socket().isBound();
    4. 
  
    
    
  
  4.     }
    5. 
  
    
    
  
    6. 
  
    
    
  
  6.     @Override
    7. 
  
    
    
  
  7.     public InetSocketAddress remoteAddress() {
    8. 
  
    
    
  
  8.         return null;
    9. 
  
    
    
  
  9.     }
    10. 
  
    
    
  
    11. 
  
    
    
  
  11.     @Override
    12. 
  
    
    
  
  12.     protected ServerSocketChannel javaChannel() {
    13. 
  
    
    
  
  13.         return (ServerSocketChannel) super.javaChannel();
    14. 
  
    
    
  
  14.     }
    15. 
  
    
    
  
    16. 
  
    
    
  
  16.     @Override
    17. 
  
    
    
  
  17.     protected SocketAddress localAddress0() {
    18. 
  
    
    
  
  18.         return SocketUtils.localSocketAddress(javaChannel().socket());
    19. 
  
    
    
  
  19.     }
    20. 
  
    
    
  
    21. 
  
    
    
  
  21.     @Override
    22. 
  
    
    
  
  22.     protected void doBind(SocketAddress localAddress) throws Exception {
    23. 
  
    
    
  
  23.         if (PlatformDependent.javaVersion() >= 7) {
    24. 
  
    
    
  
  24.             javaChannel().bind(localAddress, config.getBacklog());
    25. 
  
    
    
  
  25.         } else {
    26. 
  
    
    
  
  26.             javaChannel().socket().bind(localAddress, config.getBacklog());
    27. 
  
    
    
  
  27.         }
    28. 
  
    
    
  
  28.     }
    29.

doBind方法一来运行时JAVA的版本，如果大于7就调用ServerSocketChannel的bind方法，否则调用ServerSocket的bind方法。

 
   
   
 
    
  
    
    
  
  1.     @Override
    2. 
  
    
    
  
  2.     protected int doReadMessages(List<Object> buf) throws Exception {
    3. 
  
    
    
  
  3.         SocketChannel ch = SocketUtils.accept(javaChannel());
    4. 
  
    
    
  
    5. 
  
    
    
  
  5.         try {
    6. 
  
    
    
  
  6.             if (ch != null) {
    7. 
  
    
    
  
  7.                 buf.add(new NioSocketChannel(this, ch));
    8. 
  
    
    
  
  8.                 return 1;
    9. 
  
    
    
  
  9.             }
    10. 
  
    
    
  
  10.         } catch (Throwable t) {
    11. 
  
    
    
  
  11.             logger.warn("Failed to create a new channel from an accepted socket.", t);
    12. 
  
    
    
  
    13. 
  
    
    
  
  13.             try {
    14. 
  
    
    
  
  14.                 ch.close();
    15. 
  
    
    
  
  15.             } catch (Throwable t2) {
    16. 
  
    
    
  
  16.                 logger.warn("Failed to close a socket.", t2);
    17. 
  
    
    
  
  17.             }
    18. 
  
    
    
  
  18.         }
    19. 
  
    
    
  
    20. 
  
    
    
  
  20.         return 0;
    21. 
  
    
    
  
  21.     }
    22.

首先通过SocketUtils.accept来接受新的连接，如果新的连接不为空，则借助ServerSocketChannel和新接受的SocketChannel来创建一个NioSocketChannel， 并将NioSocketChannel添加到List buf，然后返回1，表示服务端接受消息成功。


对于NioServerSocketChannel来说，它的读取操作就是接收客户端的连接，创建NioSocketChannel对象。

无关API

一些方法是与客户端Channel相关的，因此，对于服务端Channel无须实现，如果这些方法被误调，则返回UnsupportedOperationException异常。

 
   
   
 
    
  
    
    
  
  1.     @Override
    2. 
  
    
    
  
  2.     protected boolean doConnect(
    3. 
  
    
    
  
  3.             SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
    4. 
  
    
    
  
  4.         throw new UnsupportedOperationException();
    5. 
  
    
    
  
  5.     }
    6. 
  
    
    
  
    7. 
  
    
    
  
  7.     @Override
    8. 
  
    
    
  
  8.     protected void doFinishConnect() throws Exception {
    9. 
  
    
    
  
  9.         throw new UnsupportedOperationException();
    10. 
  
    
    
  
  10.     }
    11. 
  
    
    
  
    12. 
  
    
    
  
  12.     @Override
    13. 
  
    
    
  
  13.     protected SocketAddress remoteAddress0() {
    14. 
  
    
    
  
  14.         return null;
    15. 
  
    
    
  
  15.     }
    16. 
  
    
    
  
    17. 
  
    
    
  
  17.     @Override
    18. 
  
    
    
  
  18.     protected void doDisconnect() throws Exception {
    19. 
  
    
    
  
  19.         throw new UnsupportedOperationException();
    20. 
  
    
    
  
  20.     }
    21. 
  
    
    
  
    22. 
  
    
    
  
  22.     @Override
    23. 
  
    
    
  
  23.     protected boolean doWriteMessage(Object msg, ChannelOutboundBuffer in) throws Exception {
    24. 
  
    
    
  
  24.         throw new UnsupportedOperationException();
    25. 
  
    
    
  
  25.     }
    26.

NioSocketChannel

连接操作

 
   
   
 
    
  
    
    
  
  1.     @Override
    2. 
  
    
    
  
  2.     protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
    3. 
  
    
    
  
  3.         if (localAddress != null) {
    4. 
  
    
    
  
  4.             doBind0(localAddress);
    5. 
  
    
    
  
  5.         }
    6. 
  
    
    
  
    7. 
  
    
    
  
  7.         boolean success = false;
    8. 
  
    
    
  
  8.         try {
    9. 
  
    
    
  
  9.             boolean connected = SocketUtils.connect(javaChannel(), remoteAddress);
    10. 
  
    
    
  
  10.             if (!connected) {
    11. 
  
    
    
  
  11.                 selectionKey().interestOps(SelectionKey.OP_CONNECT);
    12. 
  
    
    
  
  12.             }
    13. 
  
    
    
  
  13.             success = true;
    14. 
  
    
    
  
  14.             return connected;
    15. 
  
    
    
  
  15.         } finally {
    16. 
  
    
    
  
  16.             if (!success) {
    17. 
  
    
    
  
  17.                 doClose();
    18. 
  
    
    
  
  18.             }
    19. 
  
    
    
  
  19.         }
    20. 
  
    
    
  
  20.     }
    21.

  1. 连接成功，返回true

  2. 暂时没有连接上，服务端没有返回ACK应答，连接结果不确定，返回false

  3. 连接失败，直接抛出I/O异常

如果是结果2，需要将NioSocketChannel的SelectionKey设置为OP_CONNECT，监听连接网络操作位。如果抛出了I/O异常，说明客户端的TCP握手请求直接 被RESET或者被拒绝，此时需要调用doClose()关闭客户端连接

 
   
   
 
    
  
    
    
  
  1.     protected void doClose() throws Exception {
    2. 
  
    
    
  
  2.         super.doClose();
    3. 
  
    
    
  
  3.         javaChannel().close();
    4. 
  
    
    
  
  4.     }
    5.

写半包

 
   
   
 
    
  
    
    
  
  1.     @Override
    2. 
  
    
    
  
  2.     protected void doWrite(ChannelOutboundBuffer in) throws Exception {
    3. 
  
    
    
  
  3.         SocketChannel ch = javaChannel();
    4. 
  
    
    
  
  4.         int writeSpinCount = config().getWriteSpinCount();
    5. 
  
    
    
  
  5.         do {
    6. 
  
    
    
  
  6.             if (in.isEmpty()) {
    7. 
  
    
    
  
  7.                 // All written so clear OP_WRITE
    8. 
  
    
    
  
  8.                 clearOpWrite();
    9. 
  
    
    
  
  9.                 // Directly return here so incompleteWrite(...) is not called.
    10. 
  
    
    
  
  10.                 return;
    11. 
  
    
    
  
  11.             }
    12. 
  
    
    
  
    13. 
  
    
    
  
  13.             // Ensure the pending writes are made of ByteBufs only.
    14. 
  
    
    
  
  14.             int maxBytesPerGatheringWrite = ((NioSocketChannelConfig) config).getMaxBytesPerGatheringWrite();
    15. 
  
    
    
  
  15.             ByteBuffer[] nioBuffers = in.nioBuffers(1024, maxBytesPerGatheringWrite);
    16. 
  
    
    
  
  16.             int nioBufferCnt = in.nioBufferCount();
    17. 
  
    
    
  
    18. 
  
    
    
  
  18.             // Always us nioBuffers() to workaround data-corruption.
    19. 
  
    
    
  
  19.             // See https://github.com/netty/netty/issues/2761
    20. 
  
    
    
  
  20.             switch (nioBufferCnt) {
    21. 
  
    
    
  
  21.                 case 0:
    22. 
  
    
    
  
  22.                     // We have something else beside ByteBuffers to write so fallback to normal writes.
    23. 
  
    
    
  
  23.                     writeSpinCount -= doWrite0(in);
    24. 
  
    
    
  
  24.                     break;
    25. 
  
    
    
  
  25.                 case 1: {
    26. 
  
    
    
  
  26.                     // Only one ByteBuf so use non-gathering write
    27. 
  
    
    
  
  27.                     // Zero length buffers are not added to nioBuffers by ChannelOutboundBuffer, so there is no need
    28. 
  
    
    
  
  28.                     // to check if the total size of all the buffers is non-zero.
    29. 
  
    
    
  
  29.                     ByteBuffer buffer = nioBuffers[0];
    30. 
  
    
    
  
  30.                     int attemptedBytes = buffer.remaining();
    31. 
  
    
    
  
  31.                     final int localWrittenBytes = ch.write(buffer);
    32. 
  
    
    
  
  32.                     if (localWrittenBytes <= 0) {
    33. 
  
    
    
  
  33.                         incompleteWrite(true);
    34. 
  
    
    
  
  34.                         return;
    35. 
  
    
    
  
  35.                     }
    36. 
  
    
    
  
  36.                     adjustMaxBytesPerGatheringWrite(attemptedBytes, localWrittenBytes, maxBytesPerGatheringWrite);
    37. 
  
    
    
  
  37.                     in.removeBytes(localWrittenBytes);
    38. 
  
    
    
  
  38.                     --writeSpinCount;
    39. 
  
    
    
  
  39.                     break;
    40. 
  
    
    
  
  40.                 }
    41. 
  
    
    
  
  41.                 default: {
    42. 
  
    
    
  
  42.                     // Zero length buffers are not added to nioBuffers by ChannelOutboundBuffer, so there is no need
    43. 
  
    
    
  
  43.                     // to check if the total size of all the buffers is non-zero.
    44. 
  
    
    
  
  44.                     // We limit the max amount to int above so cast is safe
    45. 
  
    
    
  
  45.                     long attemptedBytes = in.nioBufferSize();
    46. 
  
    
    
  
  46.                     final long localWrittenBytes = ch.write(nioBuffers, 0, nioBufferCnt);
    47. 
  
    
    
  
  47.                     if (localWrittenBytes <= 0) {
    48. 
  
    
    
  
  48.                         incompleteWrite(true);
    49. 
  
    
    
  
  49.                         return;
    50. 
  
    
    
  
  50.                     }
    51. 
  
    
    
  
  51.                     // Casting to int is safe because we limit the total amount of data in the nioBuffers to int above.
    52. 
  
    
    
  
  52.                     adjustMaxBytesPerGatheringWrite((int) attemptedBytes, (int) localWrittenBytes,
    53. 
  
    
    
  
  53.                             maxBytesPerGatheringWrite);
    54. 
  
    
    
  
  54.                     in.removeBytes(localWrittenBytes);
    55. 
  
    
    
  
  55.                     --writeSpinCount;
    56. 
  
    
    
  
  56.                     break;
    57. 
  
    
    
  
  57.                 }
    58. 
  
    
    
  
  58.             }
    59. 
  
    
    
  
  59.         } while (writeSpinCount > 0);
    60. 
  
    
    
  
    61. 
  
    
    
  
  61.         incompleteWrite(writeSpinCount < 0);
    62. 
  
    
    
  
  62.     }
    63.

首先判断ChannelOutboundBuffer消息环形数组中是否有待发送的消息，如果没有，直接清除写操作位然后返回。从消息环形数组中获取可发送的ByteBuffer 数组以及可发送的数量，如果消息只有一个，直接取第一个消息，将消息写入Channel，如果写入的字节数小于等于0，设置网络监听位为写操作位，然后后返回。 如果消息的数量大于大于1，就先取出可发送数组的总字节数。

读写操作

 
   
   
 
    
  
    
    
  
  1.     @Override
    2. 
  
    
    
  
  2.     protected int doReadBytes(ByteBuf byteBuf) throws Exception {
    3. 
  
    
    
  
  3.         final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    4. 
  
    
    
  
  4.         allocHandle.attemptedBytesRead(byteBuf.writableBytes());
    5. 
  
    
    
  
  5.         return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
    6. 
  
    
    
  
  6.     }
    7.

首先通过RecvByteBufAllocator.Handle设置从NioSocketChannel读取的字节数为ByteBuf可写的字节数，然后调用ByteBuf的writeBytes从Channel 中读取指定长度的字节。

以上是关于Netty Channel的主要内容，如果未能解决你的问题，请参考以下文章

Netty入门——组件（Channel）一

Netty入门——组件（Channel）二

netty源码走读(服务端Channel创建流程)

netty代码解析

netty优雅关闭channel通道

netty集群channel怎么跨集群