详尽Netty(三):Channel

Dear 丶 2023-02-22 13:44 23阅读 0赞

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

![架构殿堂][20200415234301759.jpg]

## 概念 ##

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

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

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

其UML图：  
![在这里插入图片描述][20200704003738968.png]

## 分类 ##

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

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

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

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

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

所有方法,如图：

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

## 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，其可以随后对它们做出响应。

状态的切换：

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

## 相关源码 ##

### 1. AbstractChannel ###

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel { 
        // 父 Channel（NioServerSocketChannel 是没有父channel的）
        private final Channel parent;
        // Channel 唯一ID
        private 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所注册的eventLoop
        private 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 上返回的 selectorKey
        volatile SelectionKey selectionKey;
        // 是否还有未读的数据
        boolean readPending;
        private final Runnable clearReadPendingRunnable = new Runnable() { 
            @Override
            public 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()** 读之前的准备

@Override
        protected void doBeginRead() throws Exception { 
            // Channel.read() or ChannelHandlerContext.read() was called
            final 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线程假死的状态.

@Override
        protected void doWrite(ChannelOutboundBuffer in) throws Exception { 
            int writeSpinCount = -1; // 写自选次数
    
            boolean setOpWrite = false;  //写操作位为0
            for (;;) { 
                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 meantime
                Runnable flushTask = this.flushTask;
                if (flushTask == null) { 
                    flushTask = this.flushTask = new Runnable() { 
                        @Override
                        public void run() { 
                            flush();
                        }
                    };
                }
                eventLoop().execute(flushTask);
            }
        }

### 4.AbstractNioMessageChannel ###

doWrite操作

@Override
        protected 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对象。

![架构殿堂][20200415234301759.jpg]

[20200415234301759.jpg]: https://img-blog.csdnimg.cn/20200415234301759.jpg
[20200704003738968.png]: https://img-blog.csdnimg.cn/20200704003738968.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppbnhpbnhpbjEzMTQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200704003757162.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppbnhpbnhpbjEzMTQ=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppbnhpbnhpbjEzMTQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200704003811222.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppbnhpbnhpbjEzMTQ=,size_16,color_FFFFFF,t_70