详尽Netty(三):Channel

如果大家对java架构相关感兴趣，可以关注下面公众号，会持续更新java基础面试题, netty, spring boot,spring cloud等系列文章，一系列干货随时送达, 超神之路从此展开, BTAJ不再是梦想!

概念

Channel 是java nio的一个基本构造。

它代表一个到实体(如一个硬件设备，一个文件、一个网络套接字或者一个能够之行一个或者多个不同的I/O操作的程序组件)的开放链接，如读操作和写操作。

可以把Channel 看做是传送(入站)或者传出(出站)数据的载体。可以被打开或者被关闭，链接或者断开连接。

其UML图：

分类

Channel：是对网络Socket的封装，抽象了网络I/O的读、写、连接与绑定。

AbstractChannel：实现了Channel接口的大部分功能，一次连接用到的Channel、ChannelId、eventLoop、pipelIne、unsafe都会保存在这里。

AbstractNioChannel：通过select的方式对读写事件进行监听。

客户端Channel：主要注册read与write事件，关注于具体数据的读写。

服务端Channel：主要注册accept事件，关注于具体连接的接入，这也是与客户端Channel的read事件最主要的区别。

所有方法,如图：

Channel重要方法和参数

eventLoop：返回分配给Channel 的EventLoop
pipeline：返回分配给Channel 的ChannelPipeline
isActive：如果Channel 是活动的，则返回true。活动的意义可能依赖于底层的传输。例如，一个Socket 传输一旦连接到了远程节点便是活动的，而一个Datagram 传输一旦被打开便是活动的。
localAddress：返回本地的SokcetAddress
remoteAddress：返回远程的SocketAddress
write：将数据写到远程节点。这个数据将被传递给ChannelPipeline，并且排队直到它被冲刷
flush：将之前已写的数据冲刷到底层传输，如一个Socket
writeAndFlush：一个简便的方法，等同于调用write()并接着调用flush()

生命周期状态

ChannelUnregistered ：Channel 已经被创建，但还未注册到EventLoop
ChannelRegistered ：Channel 已经被注册到了EventLoop
ChannelActive ：Channel 处于活动状态（已经连接到它的远程节点）。它现在可以接收和发送数据了
ChannelInactive ：Channel 没有连接到远程节点
当这些状态发生改变时，将会生成对应的事件。这些事件将会被转发给ChannelPipeline 中的ChannelHandler，其可以随后对它们做出响应。

状态的切换：

相关源码

1. AbstractChannel

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {// 父 Channel（NioServerSocketChannel 是没有父channel的）private final Channel parent;// Channel 唯一IDprivate final ChannelId id;// Unsafe 对象，封装 ByteBuf 的读写操作private final Unsafe unsafe;// 关联的 Pipeline 对象private final DefaultChannelPipeline pipeline;private final VoidChannelPromise unsafeVoidPromise = new VoidChannelPromise(this, false);private final CloseFuture closeFuture = new CloseFuture(this);// 本地地址private volatile SocketAddress localAddress;//远端地址private volatile SocketAddress remoteAddress;// EventLoop 封装的 Selector,channel所注册的eventLoopprivate volatile EventLoop eventLoop;// 是否注册private volatile boolean registered;private boolean closeInitiated;/** Cache for the string representation of this channel */private boolean strValActive;private String strVal;

构造函数

protected AbstractChannel(Channel parent, ChannelId id) {this.parent = parent;this.id = id;unsafe = newUnsafe();pipeline = newChannelPipeline();
}// Unsafe 实现交给子类实现
protected abstract AbstractUnsafe newUnsafe();// 创建 DefaultChannelPipeline 对象
protected DefaultChannelPipeline newChannelPipeline() {return new DefaultChannelPipeline(this);
}

说明

1.Unsafe类里实现了具体的连接与写数据。比如：网络的读，写，链路关闭，发起连接等。之所以命名为unsafe是不希望外部使用，并非是不安全的。

2.DefaultChannelPipeline 只是一个 Handler 的容器，也可以理解为一个Handler链，具体的逻辑由Handler处理，而每个Handler都会分配一个EventLoop，最终的请求还是要EventLoop来执行，而EventLoop中又调用Channel中的内部类Unsafe对应的方法。
新建一个channel会自动创建一个ChannelPipeline。

3.这里创建 DefaultChannelPipeline，构造中传入当前的 Channel，而读写数据都是在 ChannelPipeline 中进行的，ChannelPipeline 进行读写数据又委托给 Channel 中的 Unsafe 进行操作。

2.AbstractNioChannel

public abstract class AbstractNioChannel extends AbstractChannel {// 抽象了 SocketChannel 和 ServerSocketChannel 的公共的父类//Socketchannle和ServerSocketChannel的公共操作类，用来设置SelectableChannel相关参数和IO操作private final SelectableChannel ch;// SelectionKey.OP_READ 读事件protected final int readInterestOp;// 注册到 selector 上返回的 selectorKeyvolatile SelectionKey selectionKey;// 是否还有未读的数据boolean readPending;private final Runnable clearReadPendingRunnable = new Runnable() {@Overridepublic void run() {clearReadPending0();}};/*** The future of the current connection attempt.  If not null, subsequent* connection attempts will fail.*/// 连接操作的结果private ChannelPromise connectPromise;// 连接超时定时任务private ScheduledFuture<?> connectTimeoutFuture;// 客户端地址private SocketAddress requestedRemoteAddress;....

核心方法：

  //核心操作，注册操作//1) 如果当前注册返回的selectionKey已经被取消，则抛出CancelledKeyException异常，捕获该异常进行处理。
//2) 如果是第一次处理该异常，调用多路复用器的selectNow()方法将已经取消的selectionKey从多路复用器中删除掉。操作成功之后，将selected置为true， 说明之前失效的selectionKey已经被删除掉。继续发起下一次注册操作，如果成功则退出,
//3) 如果仍然发生CancelledKeyException异常,说明我们无法删除已经被取消的selectionKey,按照JDK的API说明，这种意外不应该发生。如果发生这种问题，则说明可能NIO的相关类库存在不可恢复的BUG,直接抛出CancelledKeyException异常到上层进行统一处理。protected void doRegister() throws Exception {boolean selected = false;for (;;) {try {selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);return;} catch (CancelledKeyException e) {if (!selected) {// Force the Selector to select now as the "canceled" SelectionKey may still be// cached and not removed because no Select.select(..) operation was called yet.eventLoop().selectNow();selected = true;} else {// We forced a select operation on the selector before but the SelectionKey is still cached// for whatever reason. JDK bug ?throw e;}}}}}

SelectionKey 常量值

public abstract class SelectionKey {public static final int OP_READ = 1 << 0;   //读操作位public static final int OP_WRITE = 1 << 2;  //写操作位public static final int OP_CONNECT = 1 << 3;  //客户端连接操作位public static final int OP_ACCEPT = 1 << 4;  //服务端接受连接操作位//如果注册的操作位为0表示只是完成注册功能，说明对任何事件都不感兴趣

doBeginRead() 读之前的准备

  @Overrideprotected void doBeginRead() throws Exception {// Channel.read() or ChannelHandlerContext.read() was calledfinal SelectionKey selectionKey = this.selectionKey;if (!selectionKey.isValid()) {return;   //key无效的话直接返回}readPending = true;  //表示读pending中final int interestOps = selectionKey.interestOps();if ((interestOps & readInterestOp) == 0) {  //表示当前没有读操作位selectionKey.interestOps(interestOps | readInterestOp);  //设置读操作位}}//SelectionKey中定义的是否可读操作public final boolean isReadable() {return (readyOps() & OP_READ) != 0;}

3.AbstractNioByteChannel

doWrite操作

配置中设置循环次数是避免半包中数据量过大，IO线程一直尝试写操作，此时IO线程无法处理其他IO操作或者定时任务，比如新的消息或者定时任务，如果网络IO慢或者对方读取慢等造成IO线程假死的状态.

    @Overrideprotected void doWrite(ChannelOutboundBuffer in) throws Exception {int writeSpinCount = -1; //        写自选次数boolean setOpWrite = false;  //写操作位为0for (;;) {Object msg = in.current();if (msg == null) {  //从环形数组ChannelOutboundBuffer弹出一条消息，如果为null，表示消息已经发送完成，// Wrote all messages.clearOpWrite();  //清除写标志位，退出循环// Directly return here so incompleteWrite(...) is not called.return;}if (msg instanceof ByteBuf) {ByteBuf buf = (ByteBuf) msg;int readableBytes = buf.readableBytes();if (readableBytes == 0) { //如果可读字节为0，则丢弃该消息，循环处理其他消息in.remove();continue;}boolean done = false;    //消息是否全部发送完毕表示long flushedAmount = 0;  //发送的字节数量if (writeSpinCount == -1) {//如果为-1的时候从配置中获取写循环次数writeSpinCount = config().getWriteSpinCount();}for (int i = writeSpinCount - 1; i >= 0; i --) {int localFlushedAmount = doWriteBytes(buf);  //由子类实现写if (localFlushedAmount == 0) {  //这里表示本次发送字节为0，发送TCP缓冲区满了，所以此时为了避免空循环一直发送，这里就将半包写表示设置为true并退出循环setOpWrite = true;break;}//发送成功就对发送的字节计数flushedAmount += localFlushedAmount;if (!buf.isReadable()) { //如果没有可读字节，表示已经发送完毕done = true; //表示发送完成，并退出循环break;}}//通知promise当前写的进度in.progress(flushedAmount); if (done) {  //如果发送完成，移除缓冲的数据in.remove();} else {如果没有完成会调用incompleteWrite方法// Break the loop and so incompleteWrite(...) is called.break;}} else if (msg instanceof FileRegion) {  //这个是文件传输和上面类似FileRegion region = (FileRegion) msg;boolean done = region.transferred() >= region.count();if (!done) {long flushedAmount = 0;if (writeSpinCount == -1) {writeSpinCount = config().getWriteSpinCount();}for (int i = writeSpinCount - 1; i >= 0; i--) {long localFlushedAmount = doWriteFileRegion(region);if (localFlushedAmount == 0) {setOpWrite = true;break;}flushedAmount += localFlushedAmount;if (region.transferred() >= region.count()) {done = true;break;}}in.progress(flushedAmount);}if (done) {in.remove();} else {// Break the loop and so incompleteWrite(...) is called.break;}} else {// Should not reach here.throw new Error();}}//如果没有完成写看看需要做的事情incompleteWrite(setOpWrite);}//未完成写操作，看看操作protected final void incompleteWrite(boolean setOpWrite) {// Did not write completely.if (setOpWrite) {  //如果当前的写操作位true，那么当前多路复用器继续轮询处理setOpWrite();} else {  //否则重新新建一个task任务，让eventLoop后面点执行flush操作，这样其他任务才能够执行// Schedule flush again later so other tasks can be picked up in the meantimeRunnable flushTask = this.flushTask;if (flushTask == null) {flushTask = this.flushTask = new Runnable() {@Overridepublic void run() {flush();}};}eventLoop().execute(flushTask);}}

4.AbstractNioMessageChannel

doWrite操作

    @Overrideprotected void doWrite(ChannelOutboundBuffer in) throws Exception {final SelectionKey key = selectionKey();final int interestOps = key.interestOps();for (;;) {Object msg = in.current();if (msg == null) {// Wrote all messages.if ((interestOps & SelectionKey.OP_WRITE) != 0) {key.interestOps(interestOps & ~SelectionKey.OP_WRITE);}break;}try {boolean done = false;for (int i = config().getWriteSpinCount() - 1; i >= 0; i--) {if (doWriteMessage(msg, in)) {done = true;break;}}if (done) {in.remove();} else {// Did not write all messages.if ((interestOps & SelectionKey.OP_WRITE) == 0) {key.interestOps(interestOps | SelectionKey.OP_WRITE);}break;}} catch (Exception e) {if (continueOnWriteError()) {in.remove(e);} else {throw e;}}}}

AbstractNioMessageChannel 和AbstractNioByteChannel的消息发送实现比较相似，

不同之处在于:一个发送的是ByteBuf或者FileRegion，它们可以直接被发送;另一个发送的则是POJO对象。